:py:mod:`anod.store` ==================== .. py:module:: anod.store .. autoapi-nested-parse:: Anod store integration. Submodules ---------- .. toctree:: :titlesonly: :maxdepth: 1 buildinfo/index.rst component/index.rst file/index.rst interface/index.rst Package Contents ---------------- Classes ~~~~~~~ .. autoapisummary:: anod.store._Store anod.store._StoreWrite anod.store.StoreWriteOnly anod.store.StoreReadOnly anod.store.Store anod.store.StoreRW anod.store.LocalStore Attributes ~~~~~~~~~~ .. autoapisummary:: anod.store.logger anod.store.DATE_FORMAT_LENGTH .. py:data:: logger .. py:data:: DATE_FORMAT_LENGTH :value: 8 .. py:class:: _Store(db: os.PathLike[str] | str | None = None) Bases: :py:obj:`e3.anod.store.interface._StoreContextManager` A class to define the context manager interface needed by a Store class. .. py:class:: TableName Bases: :py:obj:`str`, :py:obj:`enum.Enum` str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'. .. py:attribute:: buildinfos :value: 'buildinfos' .. py:attribute:: resources :value: 'resources' .. py:attribute:: files :value: 'files' .. py:attribute:: component_files :value: 'component_files' .. py:attribute:: component_releases :value: 'component_releases' .. py:attribute:: components :value: 'components' .. py:attribute:: __str__ .. py:method:: __del__() -> None .. py:method:: __enter__() -> typing_extensions.Self Enter in a new context. This method is called when used with the "with" keyword. For example: .. code-block:: python with _StoreContextManager() as x: pass :return: Self .. py:method:: __exit__(*args: object) -> None Exit a context. This method is called when exiting a "with" context. For example: .. code-block:: python with _StoreContextManager() as x: pass # __exit__ is call here .. py:method:: close() -> None Close commit all changes and close the database connection. .. py:method:: _select(table: _Store, dynamic_where_rules: _Store, dynamic_where_values: collections.abc.Sequence[int | str | None], *, static_where_rules: collections.abc.Sequence[str] = (), order_by: str = 'id DESC', use_or_filter: bool = False) -> _Store SELECT sql wrapper. This method is used for two purpose: - Factorize some code. - Allow static checking on table name. :param table: The SQL table to read. :param dynamic_where_rules: A str sequence to filter the returning value. This sequence can contain '?' which will be replaced accordingly with data in where_values. This sequence cannot be empty. :param dynamic_where_values: A sequence of values that will replace any '?' found in dynamic_where_rules. Can be empty if dynamic_where_rules doesn't contain any '?'. :param static_where_rules: Others where rules, but without any '?' into. :param order_by: A field name to determine the select order. :param use_or_filter: If multiple where_rules are provide and if this parameter is true, use 'OR' to join all rules together, otherwise 'AND' is used. :return: A list of tuple. .. py:method:: _select_one(table: _Store, rid: _Store, *, field_name: _Store = 'id') -> _Store SELECT sql wrapper that return only one element. This wrapper search for an element with a specific ID on the database. :param table: The SQL table to read. :param rid: The id to use. :param field_name: The 'id' field name in the database. :return: A tuple of element. :raise anod.store.interface.StoreError: if no or more than one element found. .. py:method:: _tuple_to_buildinfo(req_tuple: _Store) -> e3.anod.store.buildinfo.BuildInfoDict :classmethod: Convert a tuple to a BuildInfoDict. :param req_tuple: The result of a database request that should be converted into a BuildInfoDict. :return: The `dict` representation of a `BuildInfo` object. .. py:method:: _tuple_to_resource(req_tuple: _Store) -> e3.anod.store.file.ResourceDict :classmethod: Convert a tuple to a ResourceDict. :param req_tuple: The result of a database request that should be converted into a ResourceDict. :return: The `dict` representation of a resource. .. py:method:: _tuple_to_file(req_tuple: _Store, *, resource: e3.anod.store.file.ResourceDict | None = None, buildinfo: e3.anod.store.buildinfo.BuildInfoDict | None = None, internal: bool | None = None) -> e3.anod.store.file.FileDict Convert a tuple to a FileDict. This method takes optional arguments. These arguments are others pre-calculated dict. The goal is to avoid making unnecessary request to the database. :param req_tuple: The result of a database request that should be converted into a FileDict. :param resource: The resource dict linked to this file. If None, this dict is retrieved from the database. :param buildinfo: The buildinfo dict linked to this file. If None, this dict is retrieved from the database. :param internal: None if unknown, True if the file is intern to a component, False otherwise. :return: The `dict` representation of a `File` object. .. py:method:: _tuple_list_to_buildinfo_list(req_tuples: list[_Store]) -> list[e3.anod.store.buildinfo.BuildInfoDict] Convert a list of tuples into a list of BuildInfoDict. :param req_tuples: The list of results of a database request that should be converted into a list of BuildInfoDict. :return: A list of BuildinfoDict. .. py:method:: _tuple_list_to_file_list(req_tuples: list[_Store], *, buildinfo: e3.anod.store.buildinfo.BuildInfoDict | None = None) -> list[e3.anod.store.file.FileDict] Convert a list of tuples into a list of FileDict. This method takes optional arguments. These arguments are others pre-calculated dict. The goal is to avoid making unnecessary request to the database. :param req_tuples: The list of results of a database request that should be converted into a list of FileDict. :param buildinfo: The buildinfo dict linked to these files. If None, this dict is retrieved from the database. :return: A list of FileDict. .. py:method:: _select_inner_join(table: _Store, fields: list[tuple[_Store, _Store | Literal[*]]], inner_join: _Store, on: tuple[tuple[_Store, _Store], tuple[_Store, _Store]], dynamic_where_rules: collections.abc.Sequence[tuple[_Store, _Store]], dynamic_where_values: collections.abc.Sequence[int | str], *, static_where_rules: collections.abc.Sequence[str] = (), order_by: tuple[_Store, str] | None = None, use_or_filter: bool = False) -> list[tuple[str | int | None, Ellipsis]] Run a SELECT JOIN request on the database. For example, the following call: .. code-block:: python self._select_inner_join( _Store.TableName.files, [(_Store.TableName.files, "*")], _Store.TableName.component_files, ( (_Store.TableName.files, "file_id"), (_Store.TableName.component_files, "file_id"), ), [ (_Store.TableName.component_files, "component_id"), (_Store.TableName.component_files, "kind"), ], [component_id, kind], ) Result into the following SQL request: .. code-block:: sql SELECT files.* FROM files INNER JOIN component_files ON files.file_id=component_files.file_id WHERE component_files.component_id=? AND component_files.kind=? .. note:: In the previous query, the `?` will be replaced by, in order of encounter, and . :param table: The database table name. :param fields: The list of tuples representing the database fields to retrieve. The tuple should first contain the table name of this field, and next the field itself. :param inner_join: The table name to inner join. :param on: The inner join on condition as a tuple of two tuple. This condition is restricted to only one "equal" condition and tuples inside it have the same format that the fields parameters. Logically, table names in this field should be unique. :param dynamic_where_rules: See _select.where_rules. The only difference is that here, we want a tuple sequence to be able to specify which table the field comes from. :param dynamic_where_values: See _select.where_values. :param static_where_rules: See _select.static_where_rules. :param order_by: A tuple of two element. The first element is one of the "INNER JOIN" tables and the second the fields to use. For example, if an "INNER JOIN" request between buildinfo and files is used, to sort the output depending on the buildinfo creation_date field, the value for this param should be ("buildinfos", "creation_date"). :param use_or_filter: See _select.use_or_filter. :return: A list of tuples matching the actual database request. .. py:method:: _list_component_files(kind: Literal[file, source], component_id: _Store, *, component_buildinfo: e3.anod.store.buildinfo.BuildInfoDict | None = None) -> list[e3.anod.store.file.FileDict] Retrieve the files of a component. :param kind: The file kind to retrieve. :param component_id: The component id associated to these files. :param component_buildinfo: A buildinfo dict to avoid unnecessary call to the database. :return: A list of files. .. py:method:: _list_component_attachments(component_id: _Store, *, component_buildinfo: e3.anod.store.buildinfo.BuildInfoDict | None = None) -> dict[str, e3.anod.store.file.FileDict] Retrieve the files of a component. :param component_id: The component id associated to these files. :param component_buildinfo: A buildinfo dict to avoid unnecessary call to the database. :return: An attachment dict. .. py:method:: _tuple_to_comp(req_tuple: _Store, *, releases: list[str] | None = None, files: list[e3.anod.store.file.FileDict] | None = None, sources: list[e3.anod.store.file.FileDict] | None = None, attachments: dict[str, e3.anod.store.file.FileDict] | None = None, readme: e3.anod.store.file.FileDict | None = None, buildinfo: e3.anod.store.buildinfo.BuildInfoDict | None = None) -> e3.anod.store.component.ComponentDict Convert a tuple to component. Optional parameters are only used to avoid unnecessary call to the database. :param req_tuple: The tuple to convert. :param releases: A list of releases linked to the component. :param files: A list of files. :param sources: A list of sources. :param attachments: An attachment dict. :param readme: A FileDict. :param buildinfo: A BuildInfoDict. .. py:method:: _tuple_list_to_comp_list(req_tuples: list[_Store]) -> list[e3.anod.store.component.ComponentDict] Convert a list of tuples into a list of ComponentDict. :param req_tuples: The list of tuples to convert. :return: A list a ComponentDict. .. py:class:: _StoreWrite(db: os.PathLike[str] | str | None = None) Bases: :py:obj:`_Store` A class to define the context manager interface needed by a Store class. .. py:method:: _insert_or_update(table: _Store, sql: str, values: list[str | int | None]) -> _Store Execute an INSERT or UPDATE statement. This method will take care of the returning value of this statement. If Sqlite version >= 3.35, the code will add the RETURNING keyword to the request. Otherwise, an additional select statement is done to retrieve the value inserted or updated. :param table: The table name. :param sql: The sql statement to execute. :param values: The values used to fill the potential '?' in the sql statement. :return: A tuple. .. py:method:: _insert(table: _Store, fields: _Store, values: list[str | int | None]) -> _Store Insert a new row in a table. :param table: The table name. :param fields: The fields to file in the row. :param values: The values to insert. :return: A tuple. .. py:method:: _update(table: _Store, rowid: str | int, toset: _Store, values: list[str | int | None], *, id_field: str = 'id') -> _Store Update a row in a table. :param table: The table name. :param rowid: The row id to update. :param toset: The list of row fields to update. :param values: The values to update. :param id_field: The id field to use, in case the `rowid` doesn't reference the primary key of the current table. :return: A tuple. .. py:class:: StoreWriteOnly(db: os.PathLike[str] | str | None = None) Bases: :py:obj:`_StoreWrite`, :py:obj:`e3.anod.store.interface.StoreWriteInterface` Write-only interface to the Anod store for creating and uploading resources. .. py:method:: create_thirdparty(file_info: e3.anod.store.file.FileDict) -> e3.anod.store.file.FileDict See e3.anod.store.interface.StoreWriteInterface. :param file_info: the file information dictionary to create as a thirdparty .. py:method:: _insert_to_component_files(kind: Literal[file, source, attachment], file_list: collections.abc.Sequence[tuple[str | None, e3.anod.store.file.FileDict]], component_id: str | int) -> None Insert a list of files to the component_files database table. :param kind: The kind of file to insert. :param file_list: A sequence of tuples. The first element of this tuple must be None if the kind != attachment. Otherwise, this represents the attachment name. The second element should always be a FileDict. :param component_id: The component id linked to these files. .. py:method:: submit_component(component_info: e3.anod.store.component.ComponentDict) -> e3.anod.store.component.ComponentDict See e3.anod.store.interface.StoreWriteInterface. :param component_info: the component information dictionary to submit .. py:method:: submit_file(file_info: e3.anod.store.file.FileDict) -> e3.anod.store.file.FileDict See e3.anod.store.interface.StoreWriteInterface. :param file_info: the file information dictionary to submit .. py:method:: mark_build_ready(bid: str) -> bool See e3.anod.store.interface.StoreWriteInterface. :param bid: the build ID to mark as ready .. py:method:: create_build_id(setup: str, date: str, version: str) -> e3.anod.store.buildinfo.BuildInfoDict See e3.anod.store.interface.StoreWriteInterface. :param setup: the build setup name :param date: the build date :param version: the build version .. py:method:: copy_build_id(bid: str, dest_setup: str) -> e3.anod.store.buildinfo.BuildInfoDict See e3.anod.store.interface.StoreWriteInterface. :param bid: the source build ID to copy :param dest_setup: the destination setup name .. py:method:: update_file_metadata(file_info: e3.anod.store.file.FileDict) -> e3.anod.store.file.FileDict See e3.anod.store.interface.StoreWriteInterface. :param file_info: the file information dictionary containing the updated metadata .. py:method:: _add_component_attachment(component_id: str, file_id: str, name: str) -> None See e3.anod.store.interface.StoreWriteInterface. :param component_id: the component ID :param file_id: the file ID of the attachment :param name: the attachment name .. py:method:: add_component_attachment(component_id: str, file_id: str, name: str) -> None See e3.anod.store.interface.StoreWriteInterface. :param component_id: the component ID :param file_id: the file ID of the attachment :param name: the attachment name .. py:method:: _submit_file(file_info: e3.anod.store.file.FileDict) -> e3.anod.store.file.FileDict Private method. Same as self.submit_file except that it does not commit the change. Mainly for optimization purpose. :param file_info: the file information dictionary to submit .. py:class:: StoreReadOnly(db: os.PathLike[str] | str | None = None) Bases: :py:obj:`_Store`, :py:obj:`e3.anod.store.interface.StoreReadInterface` Read-only interface to the Anod store for querying and downloading resources. .. py:method:: get_build_info(bid: str) -> e3.anod.store.buildinfo.BuildInfoDict See e3.anod.store.interface.StoreReadInterface. :param bid: the build ID .. py:method:: get_latest_build_info(setup: str, date: str | None = 'all', version: str | None = 'all', ready_only: bool = True) -> e3.anod.store.buildinfo.BuildInfoDict See e3.anod.store.interface.StoreReadInterface. :param setup: the build setup name :param date: the build date (default: "all") :param version: the build version (default: "all") :param ready_only: if True, only return builds marked as ready (default: True) .. py:method:: list_release_components(name: str, component: str = 'all', version: str = 'all', platform: str = 'all') -> list[e3.anod.store.component.ComponentDict] See e3.anod.store.interface.StoreReadInterface. :param name: the release name :param component: the component name filter (default: "all") :param version: the version filter (default: "all") :param platform: the platform filter (default: "all") .. py:method:: latest_components(setup: str, date: str | None = None, platform: str = 'all', component: str = 'all', specname: str | None = None, build_id: str = 'all') -> list[e3.anod.store.component.ComponentDict] See e3.anod.store.interface.StoreReadInterface. :param setup: the build setup name :param date: the build date filter (default: None) :param platform: the platform filter (default: "all") :param component: the component name filter (default: "all") :param specname: the spec name filter (default: None) :param build_id: the build ID filter (default: "all") .. py:method:: list_components(bid: str, component: str = 'all', platform: str = 'all') -> list[e3.anod.store.component.ComponentDict] See e3.anod.store.interface.StoreReadInterface. :param bid: the build ID :param component: the component name filter (default: "all") :param platform: the platform filter (default: "all") .. py:method:: get_build_data(bid: str) -> e3.anod.store.interface.BuildDataDict See e3.anod.store.interface.StoreReadInterface. :param bid: the build ID .. py:method:: get_build_info_list(date: str | None = 'all', setup: str | None = 'all', version: str | None = 'all', nb_days: int = 1) -> list[e3.anod.store.buildinfo.BuildInfoDict] See e3.anod.store.interface.StoreReadInterface.get_build_info_list. :param date: the build date filter (default: "all") :param setup: the build setup name filter (default: "all") :param version: the build version filter (default: "all") :param nb_days: number of days to look back from the date (default: 1) .. py:method:: get_source_info(name: str, bid: str, kind: str = 'source') -> e3.anod.store.file.FileDict See e3.anod.store.interface.StoreReadInterface. :param name: the source name :param bid: the build ID :param kind: the source kind (default: "source") .. py:method:: download_resource(rid: str, path: str) -> str See e3.anod.store.interface.StoreReadInterface. :param rid: the resource ID :param path: the destination path for the downloaded resource .. py:method:: latest_thirdparty(name: str, tp_id: str = 'all', rid: str = 'all') -> e3.anod.store.file.FileDict | None See e3.anod.store.interface.StoreReadInterface. :param name: the thirdparty name :param tp_id: the thirdparty ID filter (default: "all") :param rid: the resource ID filter (default: "all") .. py:method:: bulk_query(queries: list[dict[str, Any]]) -> list[dict[str, Any]] See e3.anod.store.interface.StoreReadInterface. :param queries: list of query dictionaries to execute .. py:method:: _get_file(name: str | None = None, fid: str | None = None, kind: str | None = None, bid: str | None = None, rid: str | None = None, possibly_empty: bool = False) -> e3.anod.store.file.FileDict | None Retrieve a file. :param name: The file name :param fid: The file id :param kind: The file kind :param bid: The file buildinfo id :param rid: The id of the resource of the file :param possibly_empty: If true the method can return an empty result. Otherwise, if no result is available, this method raises a StoreError. :raise anod.store.interface.StoreError: if possibly_empty is False and no result is found. .. py:method:: _get_buildinfo(dynamic_where_rules: collections.abc.Sequence[_Store], dynamic_where_values: collections.abc.Sequence[str], *, static_where_rules: collections.abc.Sequence[str] = (), only_one: bool = True) -> e3.anod.store.buildinfo.BuildInfoDict Retrieve a buildinfo. :param dynamic_where_rules: See _Store._select :param dynamic_where_values: See _Store._select :param static_where_rules: See _Store._select :param only_one: If true, no multiple results is allowed. :raise anod.store.interface.StoreError: Raises a StoreError if no result is found or if only_one is true and more than one result is available. .. py:class:: Store(db: os.PathLike[str] | str | None = None) Bases: :py:obj:`StoreReadOnly` Read-only Anod store (alias for StoreReadOnly). .. py:class:: StoreRW(db: os.PathLike[str] | str | None = None) Bases: :py:obj:`e3.anod.store.interface.StoreRWInterface`, :py:obj:`Store`, :py:obj:`StoreWriteOnly` Read-write interface to the Anod store. .. py:class:: LocalStore(db: os.PathLike[str] | str | None = None, online_store: e3.anod.store.interface.StoreReadInterface | e3.anod.store.interface.StoreRWInterface | None = None) Bases: :py:obj:`StoreRW`, :py:obj:`e3.anod.store.interface.LocalStoreInterface` Local cache-backed store with optional online store fallback. .. py:method:: download_resource(resource_id: str, path: str, *, online_store: e3.anod.store.interface.StoreReadInterface | None = None) -> str Interface method implementation. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.download_resource` :param resource_id: the resource ID to download :param path: the destination path for the downloaded resource :param online_store: optional online store interface to use for download .. py:method:: _raw_add_build_info(build_info: e3.anod.store.buildinfo.BuildInfoDict) -> bool Private interface method implementation. This function will not commit the change to the database automatically. Return True if something should be committed. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.raw_add_build_info` :param build_info: the build information dictionary to add :return: True if some change should be committed to the database, False otherwise. .. py:method:: raw_add_build_info(build_info: e3.anod.store.buildinfo.BuildInfoDict) -> None Interface method implementation. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.raw_add_build_info` :param build_info: the build information dictionary to add .. py:method:: add_build_info_from_store(from_store: e3.anod.store.interface.StoreReadInterface, bid: str) -> None Interface method implementation. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.add_build_info_from_store` :param from_store: the store to retrieve the build info from :param bid: the build ID to add .. py:method:: _raw_add_file(file_info: e3.anod.store.file.FileDict) -> bool Private interface method implementation. This function will not commit the change to the database automatically. Return True if something should be committed. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.raw_add_file` :param file_info: the file information dictionary to add :return: True if some change should be committed to the database, False otherwise. .. py:method:: raw_add_file(file_info: e3.anod.store.file.FileDict) -> None Interface method implementation. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.raw_add_file` :param file_info: the file information dictionary to add .. py:method:: add_source_from_store(from_store: e3.anod.store.interface.StoreReadInterface, name: str, bid: str | None = None, setup: str | None = None, date: str = 'all', kind: Literal[source, thirdparty] = 'source') -> None Interface method implementation. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.add_source_from_store` :param from_store: the store to retrieve the source from :param name: the source name :param bid: the build ID (default: None) :param setup: the build setup name (default: None) :param date: the build date (default: "all") :param kind: the source kind (default: "source") .. py:method:: _raw_add_component(component_info: e3.anod.store.component.ComponentDict) -> bool Private interface method implementation. This function will not commit the change to the database automatically. Return True if something should be committed. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.raw_add_component` :param component_info: the component information dictionary to add :return: True if some change should be committed to the database, False otherwise. .. py:method:: raw_add_component(component_info: e3.anod.store.component.ComponentDict) -> None Interface method implementation. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.raw_add_component` :param component_info: the component information dictionary to add .. py:method:: add_component_from_store(from_store: e3.anod.store.interface.StoreReadInterface, setup: str, name: str = 'all', platform: str = 'all', date: str | None = None, specname: str | None = None) -> None Interface method implementation. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.add_component_from_store` :param from_store: the store to retrieve the component from :param setup: the build setup name :param name: the component name (default: "all") :param platform: the platform (default: "all") :param date: the build date (default: None) :param specname: the spec name (default: None) .. py:method:: save(filename: pathlib.Path | str | None = None) -> None Interface method implementation. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.save` :param filename: optional path to save the database to (default: None) .. py:method:: bulk_update_from_store(from_store: e3.anod.store.interface.StoreReadInterface, queries: list[dict[str, Any]]) -> list[dict[str, Any]] Interface method implementation. .. seealso:: :py:meth:`e3.anod.store.interface.LocalStore.bulk_update_from_store` :param from_store: the store to retrieve data from :param queries: list of query dictionaries to execute