# -*- coding: utf-8 -*- ########################################################################### # Copyright (c), The AiiDA team. All rights reserved. # # This file is part of the AiiDA code. # # # # The code is hosted on GitHub at https://github.com/aiidateam/aiida-core # # For further information on the license, see the LICENSE.txt file # # For further information please visit http://www.aiida.net # ########################################################################### """Class that represents the repository of a `Node` instance. .. deprecated:: 1.4.0 This module has been deprecated and will be removed in `v2.0.0`. """ import os import warnings from aiida.common import exceptions from aiida.common.folders import RepositoryFolder, SandboxFolder from aiida.common.warnings import AiidaDeprecationWarning from aiida.repository import File, FileType [docs]class Repository: """Class that represents the repository of a `Node` instance. .. deprecated:: 1.4.0 This class has been deprecated and will be removed in `v2.0.0`. """ # Name to be used for the Repository section _section_name = 'node' [docs] def __init__(self, uuid, is_stored, base_path=None): self._is_stored = is_stored self._base_path = base_path self._temp_folder = None self._repo_folder = RepositoryFolder(section=self._section_name, uuid=uuid) [docs] def __del__(self): """Clean the sandboxfolder if it was instantiated.""" if getattr(self, '_temp_folder', None) is not None: self._temp_folder.erase() [docs] def validate_mutability(self): """Raise if the repository is immutable. :raises aiida.common.ModificationNotAllowed: if repository is marked as immutable because the corresponding node is stored """ if self._is_stored: raise exceptions.ModificationNotAllowed('cannot modify the repository after the node has been stored') [docs] @staticmethod def validate_object_key(key): """Validate the key of an object. :param key: an object key in the repository :raises ValueError: if the key is not a valid object key """ if key and os.path.isabs(key): raise ValueError('the key must be a relative path') [docs] def list_objects(self, key=None): """Return a list of the objects contained in this repository, optionally in the given sub directory. :param key: fully qualified identifier for the object within the repository :return: a list of `File` named tuples representing the objects present in directory with the given key """ folder = self._get_base_folder() if key: folder = folder.get_subfolder(key) objects = [] for filename in folder.get_content_list(): if os.path.isdir(os.path.join(folder.abspath, filename)): objects.append(File(filename, FileType.DIRECTORY)) else: objects.append(File(filename, FileType.FILE)) return sorted(objects, key=lambda x: x.name) [docs] def list_object_names(self, key=None): """Return a list of the object names contained in this repository, optionally in the given sub directory. :param key: fully qualified identifier for the object within the repository :return: a list of `File` named tuples representing the objects present in directory with the given key """ return [entry.name for entry in self.list_objects(key)] [docs] def open(self, key, mode='r'): """Open a file handle to an object stored under the given key. :param key: fully qualified identifier for the object within the repository :param mode: the mode under which to open the handle """ return open(self._get_base_folder().get_abs_path(key), mode=mode) [docs] def get_object(self, key): """Return the object identified by key. :param key: fully qualified identifier for the object within the repository :return: a `File` named tuple representing the object located at key :raises IOError: if no object with the given key exists """ self.validate_object_key(key) try: directory, filename = key.rsplit(os.sep, 1) except ValueError: directory, filename = None, key folder = self._get_base_folder() if directory: folder = folder.get_subfolder(directory) filepath = os.path.join(folder.abspath, filename) if os.path.isdir(filepath): return File(filename, FileType.DIRECTORY) if os.path.isfile(filepath): return File(filename, FileType.FILE) raise IOError(f'object {key} does not exist') [docs] def get_object_content(self, key, mode='r'): """Return the content of a object identified by key. :param key: fully qualified identifier for the object within the repository :param mode: the mode under which to open the handle """ with self.open(key, mode=mode) as handle: return handle.read() [docs] def put_object_from_tree(self, path, key=None, contents_only=True, force=False): """Store a new object under `key` with the contents of the directory located at `path` on this file system. .. warning:: If the repository belongs to a stored node, a `ModificationNotAllowed` exception will be raised. This check can be avoided by using the `force` flag, but this should be used with extreme caution! :param path: absolute path of directory whose contents to copy to the repository :param key: fully qualified identifier for the object within the repository :param contents_only: boolean, if True, omit the top level directory of the path and only copy its contents. :param force: boolean, if True, will skip the mutability check :raises aiida.common.ModificationNotAllowed: if repository is immutable and `force=False` """ if not force: self.validate_mutability() self.validate_object_key(key) if not os.path.isabs(path): raise ValueError('the `path` must be an absolute path') folder = self._get_base_folder() if key: folder = folder.get_subfolder(key, create=True) if contents_only: for entry in os.listdir(path): folder.insert_path(os.path.join(path, entry)) else: folder.insert_path(path) [docs] def put_object_from_file(self, path, key, mode=None, encoding=None, force=False): """Store a new object under `key` with contents of the file located at `path` on this file system. .. warning:: If the repository belongs to a stored node, a `ModificationNotAllowed` exception will be raised. This check can be avoided by using the `force` flag, but this should be used with extreme caution! :param path: absolute path of file whose contents to copy to the repository :param key: fully qualified identifier for the object within the repository :param mode: the file mode with which the object will be written Deprecated: will be removed in `v2.0.0` :param encoding: the file encoding with which the object will be written Deprecated: will be removed in `v2.0.0` :param force: boolean, if True, will skip the mutability check :raises aiida.common.ModificationNotAllowed: if repository is immutable and `force=False` """ # pylint: disable=unused-argument,no-member # Note that the defaults of `mode` and `encoding` had to be change to `None` from `w` and `utf-8` resptively, in # order to detect when they were being passed such that the deprecation warning can be emitted. The defaults did # not make sense and so ignoring them is justified, since the side-effect of this function, a file being copied, # will continue working the same. if mode is not None: warnings.warn('the `mode` argument is deprecated and will be removed in `v2.0.0`', AiidaDeprecationWarning) if encoding is not None: warnings.warn( 'the `encoding` argument is deprecated and will be removed in `v2.0.0`', AiidaDeprecationWarning ) if not force: self.validate_mutability() self.validate_object_key(key) with open(path, mode='rb') as handle: self.put_object_from_filelike(handle, key, mode='wb', encoding=None) [docs] def put_object_from_filelike(self, handle, key, mode='w', encoding='utf8', force=False): """Store a new object under `key` with contents of filelike object `handle`. .. warning:: If the repository belongs to a stored node, a `ModificationNotAllowed` exception will be raised. This check can be avoided by using the `force` flag, but this should be used with extreme caution! :param handle: filelike object with the content to be stored :param key: fully qualified identifier for the object within the repository :param mode: the file mode with which the object will be written :param encoding: the file encoding with which the object will be written :param force: boolean, if True, will skip the mutability check :raises aiida.common.ModificationNotAllowed: if repository is immutable and `force=False` """ if not force: self.validate_mutability() self.validate_object_key(key) folder = self._get_base_folder() while os.sep in key: basepath, key = key.split(os.sep, 1) folder = folder.get_subfolder(basepath, create=True) folder.create_file_from_filelike(handle, key, mode=mode, encoding=encoding) [docs] def delete_object(self, key, force=False): """Delete the object from the repository. .. warning:: If the repository belongs to a stored node, a `ModificationNotAllowed` exception will be raised. This check can be avoided by using the `force` flag, but this should be used with extreme caution! :param key: fully qualified identifier for the object within the repository :param force: boolean, if True, will skip the mutability check :raises aiida.common.ModificationNotAllowed: if repository is immutable and `force=False` """ if not force: self.validate_mutability() self.validate_object_key(key) self._get_base_folder().remove_path(key) [docs] def erase(self, force=False): """Delete the repository folder. .. warning:: If the repository belongs to a stored node, a `ModificationNotAllowed` exception will be raised. This check can be avoided by using the `force` flag, but this should be used with extreme caution! :param force: boolean, if True, will skip the mutability check :raises aiida.common.ModificationNotAllowed: if repository is immutable and `force=False` """ if not force: self.validate_mutability() self._get_base_folder().erase() [docs] def store(self): """Store the contents of the sandbox folder into the repository folder.""" if self._is_stored: raise exceptions.ModificationNotAllowed('repository is already stored') self._repo_folder.replace_with_folder(self._get_temp_folder().abspath, move=True, overwrite=True) self._is_stored = True [docs] def restore(self): """Move the contents from the repository folder back into the sandbox folder.""" if not self._is_stored: raise exceptions.ModificationNotAllowed('repository is not yet stored') self._temp_folder.replace_with_folder(self._repo_folder.abspath, move=True, overwrite=True) self._is_stored = False [docs] def _get_base_folder(self): """Return the base sub folder in the repository. :return: a Folder object. """ if self._is_stored: folder = self._repo_folder else: folder = self._get_temp_folder() if self._base_path is not None: folder = folder.get_subfolder(self._base_path, reset_limit=True) folder.create() return folder [docs] def _get_temp_folder(self): """Return the temporary sandbox folder. :return: a SandboxFolder object mapping the node in the repository. """ if self._temp_folder is None: self._temp_folder = SandboxFolder() return self._temp_folder