Source code for aiida.orm.utils._repository

# -*- 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