Storage implementation using Sqlite database and disk-objectstore container.


Storage implementation using Sqlite database and disk-objectstore container.

class Profile)[source]#

Bases: PsqlDosMigrator

Storage implementation using Sqlite database and disk-objectstore container.

This storage backend is not recommended for use in production. The sqlite database is not the most performant and it does not support all the QueryBuilder functionality that is supported by the core.psql_dos storage backend. This storage is ideally suited for use cases that want to test or demo AiiDA as it requires no server but just a folder on the local filesystem.

__annotations__ = {}#
__init__(profile: Profile) None[source]#
__module__ = ''#
get_container() Container[source]#

Return the disk-object store container.


The disk-object store container configured for the repository path of the current profile.

initialise_database() None[source]#

Initialise the database.

This assumes that the database has no schema whatsoever and so the initial schema is created directly from the models at the current head version without migrating through all of them one by one.

class Profile)[source]#

Bases: PsqlDosBackend

A lightweight backend intended for demos and testing.

This backend implementation uses an Sqlite database and

class Model(**data: Any)[source]#

Bases: BaseModel

Model describing required information to configure an instance of the storage.

__abstractmethods__ = frozenset({})#
__annotations__ = {'__class_vars__': 'ClassVar[set[str]]', '__private_attributes__': 'ClassVar[dict[str, ModelPrivateAttr]]', '__pydantic_complete__': 'ClassVar[bool]', '__pydantic_core_schema__': 'ClassVar[CoreSchema]', '__pydantic_custom_init__': 'ClassVar[bool]', '__pydantic_decorators__': 'ClassVar[_decorators.DecoratorInfos]', '__pydantic_extra__': 'dict[str, Any] | None', '__pydantic_fields_set__': 'set[str]', '__pydantic_generic_metadata__': 'ClassVar[_generics.PydanticGenericMetadata]', '__pydantic_parent_namespace__': 'ClassVar[dict[str, Any] | None]', '__pydantic_post_init__': "ClassVar[None | Literal['model_post_init']]", '__pydantic_private__': 'dict[str, Any] | None', '__pydantic_root_model__': 'ClassVar[bool]', '__pydantic_serializer__': 'ClassVar[SchemaSerializer]', '__pydantic_validator__': 'ClassVar[SchemaValidator]', '__signature__': 'ClassVar[Signature]', 'filepath': 'str', 'model_computed_fields': 'ClassVar[dict[str, ComputedFieldInfo]]', 'model_config': 'ClassVar[ConfigDict]', 'model_fields': 'ClassVar[dict[str, FieldInfo]]'}#
__class_vars__: ClassVar[set[str]] = {}#
__module__ = ''#
__private_attributes__: ClassVar[dict[str, ModelPrivateAttr]] = {}#
__pydantic_complete__: ClassVar[bool] = True#
__pydantic_custom_init__: ClassVar[bool] = False#
__pydantic_decorators__: ClassVar[_decorators.DecoratorInfos] = DecoratorInfos(validators={}, field_validators={'filepath_is_absolute': Decorator(cls_ref='', cls_var_name='filepath_is_absolute', func=<bound method SqliteDosStorage.Model.filepath_is_absolute of <class ''>>, shim=None, info=FieldValidatorDecoratorInfo(fields=('filepath',), mode='after', check_fields=None))}, root_validators={}, field_serializers={}, model_serializers={}, model_validators={}, computed_fields={})#
__pydantic_extra__: dict[str, Any] | None#
__pydantic_fields_set__: set[str]#
__pydantic_generic_metadata__: ClassVar[_generics.PydanticGenericMetadata] = {'args': (), 'origin': None, 'parameters': ()}#
__pydantic_parent_namespace__: ClassVar[dict[str, Any] | None] = {'__doc__': 'A lightweight backend intended for demos and testing.\n\n    This backend implementation uses an Sqlite database and\n    ', '__module__': '', '__qualname__': 'SqliteDosStorage', 'migrator': <pydantic._internal._model_construction._PydanticWeakRef object>, 'subject': <pydantic._internal._mock_val_ser.MockValSer object>}#
__pydantic_post_init__: ClassVar[None | Literal['model_post_init']] = None#
__pydantic_private__: dict[str, Any] | None#
__pydantic_serializer__: ClassVar[SchemaSerializer] = SchemaSerializer(serializer=Model(     ModelSerializer {         class: Py(             0x000055a150276a20,         ),         serializer: Fields(             GeneralFieldsSerializer {                 fields: {                     "filepath": SerField {                         key_py: Py(                             0x00007fbec7e89900,                         ),                         alias: None,                         alias_py: None,                         serializer: Some(                             WithDefault(                                 WithDefaultSerializer {                                     default: DefaultFactory(                                         Py(                                             0x00007fbebcc76700,                                         ),                                     ),                                     serializer: Str(                                         StrSerializer,                                     ),                                 },                             ),                         ),                         required: true,                     },                 },                 computed_fields: Some(                     ComputedFields(                         [],                     ),                 ),                 mode: SimpleDict,                 extra_serializer: None,                 filter: SchemaFilter {                     include: None,                     exclude: None,                 },                 required_fields: 1,             },         ),         has_extra: false,         root_model: false,         name: "Model",     }, ), definitions=[])#
__pydantic_validator__: ClassVar[SchemaValidator] = SchemaValidator(title="Model", validator=Model(     ModelValidator {         revalidate: Never,         validator: ModelFields(             ModelFieldsValidator {                 fields: [                     Field {                         name: "filepath",                         lookup_key: Simple {                             key: "filepath",                             py_key: Py(                                 0x00007fbe8dc6da70,                             ),                             path: LookupPath(                                 [                                     S(                                         "filepath",                                         Py(                                             0x00007fbe8bf251f0,                                         ),                                     ),                                 ],                             ),                         },                         name_py: Py(                             0x00007fbec7e89900,                         ),                         validator: WithDefault(                             WithDefaultValidator {                                 default: DefaultFactory(                                     Py(                                         0x00007fbebcc76700,                                     ),                                 ),                                 on_error: Raise,                                 validator: FunctionAfter(                                     FunctionAfterValidator {                                         validator: Str(                                             StrValidator {                                                 strict: false,                                                 coerce_numbers_to_str: false,                                             },                                         ),                                         func: Py(                                             0x00007fbebcc5eac0,                                         ),                                         config: Py(                                             0x00007fbe8d07efc0,                                         ),                                         name: "function-after[filepath_is_absolute(), str]",                                         field_name: None,                                         info_arg: false,                                     },                                 ),                                 validate_default: false,                                 copy_default: false,                                 name: "default[function-after[filepath_is_absolute(), str]]",                                 undefined: Py(                                     0x00007fbec57ced60,                                 ),                             },                         ),                         frozen: false,                     },                 ],                 model_name: "Model",                 extra_behavior: Ignore,                 extras_validator: None,                 strict: false,                 from_attributes: false,                 loc_by_alias: true,             },         ),         class: Py(             0x000055a150276a20,         ),         post_init: None,         frozen: false,         custom_init: false,         root_model: false,         undefined: Py(             0x00007fbec57ced60,         ),         name: "Model",     }, ), definitions=[], cache_strings=True)#

list of weak references to the object (if defined)

_abc_impl = <_abc._abc_data object>#
filepath: str#
classmethod filepath_is_absolute(value: str) str[source]#

Return the resolved and absolute filepath.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'defer_build': True}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'filepath': FieldInfo(annotation=str, required=False, default_factory=<lambda>, title='Directory of the backend', description='Filepath of the directory in which to store data for this backend.')}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

__abstractmethods__ = frozenset({})#
__module__ = ''#
__str__() str[source]#

Return a string showing connection details for this instance.

_abc_impl = <_abc._abc_data object>#
_backup(dest: str, keep: int | None = None)[source]#

Create a backup of the storage.

  • dest – Path to where the backup will be created. Can be a path on the local file system, or a path on a remote that can be accessed over SSH in the form <user>@<host>:<path>.

  • keep – The maximum number of backups to keep. If the number of copies exceeds this number, the oldest backups are removed.

_backup_storage(manager: BackupManager, path: Path, prev_backup: Path | None = None) None[source]#

Create a backup of the sqlite database and disk-objectstore to the provided path.

  • manager – BackupManager from backup_utils containing utilities such as for calling the rsync.

  • path – Path to where the backup will be created.

  • prev_backup – Path to the previous backup. Rsync calls will be hard-linked to this path, making the backup incremental and efficient.

_default_user: 'User' | None#
static _get_mapper_from_entity(entity_type: EntityTypes, with_pk: bool)[source]#

Return the Sqlalchemy mapper and fields corresponding to the given entity.


with_pk – if True, the fields returned will include the primary key


Initialise the SQLAlchemy session factory.

Only one session factory is ever associated with a given class instance, i.e. once the instance is closed, it cannot be reopened.

The session factory, returns a session that is bound to the current thread. Multi-thread support is currently required by the REST API. Although, in the future, we may want to move the multi-thread handling to higher in the AiiDA stack.

_session_factory: scoped_session | None#
property authinfos#

Return the collection of authorisation information objects

property comments#

Return the collection of comments

property computers#

Return the collection of computers

delete() None[source]#

Delete the storage and all the data.

property filepath_container: Path#
property filepath_database: Path#
property filepath_root: Path#
get_backend_entity(model) BackendEntity[source]#

Return the backend entity that corresponds to the given Model instance.

get_container() Container[source]#
get_repository() DiskObjectStoreRepositoryBackend[source]#

Return the object repository configured for this backend.

property groups#

Return the collection of groups

classmethod initialise(profile: Profile, reset: bool = False) bool[source]#

Initialise the storage backend.

This is typically used once when a new storage backed is created. If this method returns without exceptions the storage backend is ready for use. If the backend already seems initialised, this method is a no-op.


reset – If true, destroy the backend if it already exists including all of its data before recreating and initialising it. This is useful for example for test profiles that need to be reset before or after tests having run.


True if the storage was initialised by the function call, False if it was already initialised.

property logs#

Return the collection of logs


alias of SqliteDosMigrator

property nodes#

Return the collection of nodes

query() SqliteQueryBuilder[source]#

Return an instance of a query builder implementation for this backend

property users#

Return the collection of users

classmethod version_head() str[source]#

Return the head schema version of this storage backend type.

classmethod version_profile(profile: Profile) str | None[source]#

Return the schema version of the given profile’s storage, or None for empty/uninitialised storage.


~aiida.common.exceptions.UnreachableStorage if the storage cannot be accessed