anod.store

Anod store integration.

Submodules

• anod.store.buildinfo
• anod.store.component
• anod.store.file
• anod.store.interface

Package Contents

Classes

_Store	A class to define the context manager interface needed by a Store class.
_StoreWrite	A class to define the context manager interface needed by a Store class.
StoreWriteOnly	Write-only interface to the Anod store for creating and uploading resources.
StoreReadOnly	Read-only interface to the Anod store for querying and downloading resources.
Store	Read-only Anod store (alias for StoreReadOnly).
StoreRW	Read-write interface to the Anod store.
LocalStore	Local cache-backed store with optional online store fallback.

Attributes

logger
DATE_FORMAT_LENGTH

anod.store.logger

anod.store.DATE_FORMAT_LENGTH = 8

class anod.store._Store(db: os.PathLike[str] | str | None = None)

Bases: e3.anod.store.interface._StoreContextManager

A class to define the context manager interface needed by a Store class.

class TableName

Bases: str, 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’.

buildinfos = 'buildinfos'

resources = 'resources'

files = 'files'

component_files = 'component_files'

component_releases = 'component_releases'

components = 'components'

__str__

__del__() → None

__enter__() → typing_extensions.Self

Enter in a new context.

This method is called when used with the “with” keyword. For example:

with _StoreContextManager() as x:
    pass

Returns:

Self

__exit__(*args: object) → None

Exit a context.

This method is called when exiting a “with” context. For example:

with _StoreContextManager() as x:
    pass
# __exit__ is call here

close() → None

Close commit all changes and close the database connection.

_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.

Parameters:

• table – The SQL table to read.

• 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.

• 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 ‘?’.

• static_where_rules – Others where rules, but without any ‘?’ into.

• order_by – A field name to determine the select order.

• 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.

Returns:

A list of tuple.

_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.

Parameters:

• table – The SQL table to read.

• rid – The id to use.

• field_name – The ‘id’ field name in the database.

Returns:

A tuple of element.

Raises:

anod.store.interface.StoreError – if no or more than one element found.

classmethod _tuple_to_buildinfo(req_tuple: _Store) → e3.anod.store.buildinfo.BuildInfoDict

Convert a tuple to a BuildInfoDict.

Parameters:

req_tuple – The result of a database request that should be converted into a BuildInfoDict.

Returns:

The dict representation of a BuildInfo object.

classmethod _tuple_to_resource(req_tuple: _Store) → e3.anod.store.file.ResourceDict

Convert a tuple to a ResourceDict.

Parameters:

req_tuple – The result of a database request that should be converted into a ResourceDict.

Returns:

The dict representation of a resource.

_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.

Parameters:

• req_tuple – The result of a database request that should be converted into a FileDict.

• resource – The resource dict linked to this file. If None, this dict is retrieved from the database.

• buildinfo – The buildinfo dict linked to this file. If None, this dict is retrieved from the database.

• internal – None if unknown, True if the file is intern to a component, False otherwise.

Returns:

The dict representation of a File object.

_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.

Parameters:

req_tuples – The list of results of a database request that should be converted into a list of BuildInfoDict.

Returns:

A list of BuildinfoDict.

_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.

Parameters:

• req_tuples – The list of results of a database request that should be converted into a list of FileDict.

• buildinfo – The buildinfo dict linked to these files. If None, this dict is retrieved from the database.

Returns:

A list of FileDict.

_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:

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:

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, <component_id> and <kind>.

Parameters:

• table – The database table name.

• 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.

• inner_join – The table name to inner join.

• 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.

• 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.

• dynamic_where_values – See _select.where_values.

• static_where_rules – See _select.static_where_rules.

• 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”).

• use_or_filter – See _select.use_or_filter.

Returns:

A list of tuples matching the actual database request.

_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.

Parameters:

• kind – The file kind to retrieve.

• component_id – The component id associated to these files.

• component_buildinfo – A buildinfo dict to avoid unnecessary call to the database.

Returns:

A list of files.

_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.

Parameters:

• component_id – The component id associated to these files.

• component_buildinfo – A buildinfo dict to avoid unnecessary call to the database.

Returns:

An attachment dict.

_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.

Parameters:

• req_tuple – The tuple to convert.

• releases – A list of releases linked to the component.

• files – A list of files.

• sources – A list of sources.

• attachments – An attachment dict.

• readme – A FileDict.

• buildinfo – A BuildInfoDict.

_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.

Parameters:

req_tuples – The list of tuples to convert.

Returns:

A list a ComponentDict.

class anod.store._StoreWrite(db: os.PathLike[str] | str | None = None)

Bases: _Store

A class to define the context manager interface needed by a Store class.

_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.

Parameters:

• table – The table name.

• sql – The sql statement to execute.

• values – The values used to fill the potential ‘?’ in the sql statement.

Returns:

A tuple.

_insert(table: _Store, fields: _Store, values: list[str | int | None]) → _Store

Insert a new row in a table.

Parameters:

• table – The table name.

• fields – The fields to file in the row.

• values – The values to insert.

Returns:

A tuple.

_update(table: _Store, rowid: str | int, toset: _Store, values: list[str | int | None], *, id_field: str = 'id') → _Store

Update a row in a table.

Parameters:

• table – The table name.

• rowid – The row id to update.

• toset – The list of row fields to update.

• values – The values to update.

• id_field – The id field to use, in case the rowid doesn’t reference the primary key of the current table.

Returns:

A tuple.

class anod.store.StoreWriteOnly(db: os.PathLike[str] | str | None = None)

Bases: _StoreWrite, e3.anod.store.interface.StoreWriteInterface

Write-only interface to the Anod store for creating and uploading resources.

create_thirdparty(file_info: e3.anod.store.file.FileDict) → e3.anod.store.file.FileDict

See e3.anod.store.interface.StoreWriteInterface.

Parameters:

file_info – the file information dictionary to create as a thirdparty

_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.

Parameters:

• kind – The kind of file to insert.

• 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.

• component_id – The component id linked to these files.

submit_component(component_info: e3.anod.store.component.ComponentDict) → e3.anod.store.component.ComponentDict

See e3.anod.store.interface.StoreWriteInterface.

Parameters:

component_info – the component information dictionary to submit

submit_file(file_info: e3.anod.store.file.FileDict) → e3.anod.store.file.FileDict

See e3.anod.store.interface.StoreWriteInterface.

Parameters:

file_info – the file information dictionary to submit

mark_build_ready(bid: str) → bool

See e3.anod.store.interface.StoreWriteInterface.

Parameters:

bid – the build ID to mark as ready

create_build_id(setup: str, date: str, version: str) → e3.anod.store.buildinfo.BuildInfoDict

See e3.anod.store.interface.StoreWriteInterface.

Parameters:

• setup – the build setup name

• date – the build date

• version – the build version

copy_build_id(bid: str, dest_setup: str) → e3.anod.store.buildinfo.BuildInfoDict

See e3.anod.store.interface.StoreWriteInterface.

Parameters:

• bid – the source build ID to copy

• dest_setup – the destination setup name

update_file_metadata(file_info: e3.anod.store.file.FileDict) → e3.anod.store.file.FileDict

See e3.anod.store.interface.StoreWriteInterface.

Parameters:

file_info – the file information dictionary containing the updated metadata

_add_component_attachment(component_id: str, file_id: str, name: str) → None

See e3.anod.store.interface.StoreWriteInterface.

Parameters:

• component_id – the component ID

• file_id – the file ID of the attachment

• name – the attachment name

add_component_attachment(component_id: str, file_id: str, name: str) → None

See e3.anod.store.interface.StoreWriteInterface.

Parameters:

• component_id – the component ID

• file_id – the file ID of the attachment

• name – the attachment name

_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.

Parameters:

file_info – the file information dictionary to submit

class anod.store.StoreReadOnly(db: os.PathLike[str] | str | None = None)

Bases: _Store, e3.anod.store.interface.StoreReadInterface

Read-only interface to the Anod store for querying and downloading resources.

get_build_info(bid: str) → e3.anod.store.buildinfo.BuildInfoDict

See e3.anod.store.interface.StoreReadInterface.

Parameters:

bid – the build ID

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.

Parameters:

• setup – the build setup name

• date – the build date (default: “all”)

• version – the build version (default: “all”)

• ready_only – if True, only return builds marked as ready (default: True)

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.

Parameters:

• name – the release name

• component – the component name filter (default: “all”)

• version – the version filter (default: “all”)

• platform – the platform filter (default: “all”)

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.

Parameters:

• setup – the build setup name

• date – the build date filter (default: None)

• platform – the platform filter (default: “all”)

• component – the component name filter (default: “all”)

• specname – the spec name filter (default: None)

• build_id – the build ID filter (default: “all”)

list_components(bid: str, component: str = 'all', platform: str = 'all') → list[e3.anod.store.component.ComponentDict]

See e3.anod.store.interface.StoreReadInterface.

Parameters:

• bid – the build ID

• component – the component name filter (default: “all”)

• platform – the platform filter (default: “all”)

get_build_data(bid: str) → e3.anod.store.interface.BuildDataDict

See e3.anod.store.interface.StoreReadInterface.

Parameters:

bid – the build ID

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.

Parameters:

• date – the build date filter (default: “all”)

• setup – the build setup name filter (default: “all”)

• version – the build version filter (default: “all”)

• nb_days – number of days to look back from the date (default: 1)

get_source_info(name: str, bid: str, kind: str = 'source') → e3.anod.store.file.FileDict

See e3.anod.store.interface.StoreReadInterface.

Parameters:

• name – the source name

• bid – the build ID

• kind – the source kind (default: “source”)

download_resource(rid: str, path: str) → str

See e3.anod.store.interface.StoreReadInterface.

Parameters:

• rid – the resource ID

• path – the destination path for the downloaded resource

latest_thirdparty(name: str, tp_id: str = 'all', rid: str = 'all') → e3.anod.store.file.FileDict | None

See e3.anod.store.interface.StoreReadInterface.

Parameters:

• name – the thirdparty name

• tp_id – the thirdparty ID filter (default: “all”)

• rid – the resource ID filter (default: “all”)

bulk_query(queries: list[dict[str, Any]]) → list[dict[str, Any]]

See e3.anod.store.interface.StoreReadInterface.

Parameters:

queries – list of query dictionaries to execute

_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.

Parameters:

• name – The file name

• fid – The file id

• kind – The file kind

• bid – The file buildinfo id

• rid – The id of the resource of the file

• possibly_empty – If true the method can return an empty result. Otherwise, if no result is available, this method raises a StoreError.

Raises:

anod.store.interface.StoreError – if possibly_empty is False and no result is found.

_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.

Parameters:

• dynamic_where_rules – See _Store._select

• dynamic_where_values – See _Store._select

• static_where_rules – See _Store._select

• only_one – If true, no multiple results is allowed.

Raises:

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.

class anod.store.Store(db: os.PathLike[str] | str | None = None)

Bases: StoreReadOnly

Read-only Anod store (alias for StoreReadOnly).

class anod.store.StoreRW(db: os.PathLike[str] | str | None = None)

Bases: e3.anod.store.interface.StoreRWInterface, Store, StoreWriteOnly

Read-write interface to the Anod store.

class anod.store.LocalStore(db: os.PathLike[str] | str | None = None, online_store: e3.anod.store.interface.StoreReadInterface | e3.anod.store.interface.StoreRWInterface | None = None)

Bases: StoreRW, e3.anod.store.interface.LocalStoreInterface

Local cache-backed store with optional online store fallback.

download_resource(resource_id: str, path: str, *, online_store: e3.anod.store.interface.StoreReadInterface | None = None) → str

Interface method implementation.

See also

e3.anod.store.interface.LocalStore.download_resource()

Parameters:

• resource_id – the resource ID to download

• path – the destination path for the downloaded resource

• online_store – optional online store interface to use for download

_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.

See also

e3.anod.store.interface.LocalStore.raw_add_build_info()

Parameters:

build_info – the build information dictionary to add

Returns:

True if some change should be committed to the database, False otherwise.

raw_add_build_info(build_info: e3.anod.store.buildinfo.BuildInfoDict) → None

Interface method implementation.

See also

e3.anod.store.interface.LocalStore.raw_add_build_info()

Parameters:

build_info – the build information dictionary to add

add_build_info_from_store(from_store: e3.anod.store.interface.StoreReadInterface, bid: str) → None

Interface method implementation.

See also

e3.anod.store.interface.LocalStore.add_build_info_from_store()

Parameters:

• from_store – the store to retrieve the build info from

• bid – the build ID to add

_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.

See also

e3.anod.store.interface.LocalStore.raw_add_file()

Parameters:

file_info – the file information dictionary to add

Returns:

True if some change should be committed to the database, False otherwise.

raw_add_file(file_info: e3.anod.store.file.FileDict) → None

Interface method implementation.

See also

e3.anod.store.interface.LocalStore.raw_add_file()

Parameters:

file_info – the file information dictionary to add

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.

See also

e3.anod.store.interface.LocalStore.add_source_from_store()

Parameters:

• from_store – the store to retrieve the source from

• name – the source name

• bid – the build ID (default: None)

• setup – the build setup name (default: None)

• date – the build date (default: “all”)

• kind – the source kind (default: “source”)

_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.

See also

e3.anod.store.interface.LocalStore.raw_add_component()

Parameters:

component_info – the component information dictionary to add

Returns:

True if some change should be committed to the database, False otherwise.

raw_add_component(component_info: e3.anod.store.component.ComponentDict) → None

Interface method implementation.

See also

e3.anod.store.interface.LocalStore.raw_add_component()

Parameters:

component_info – the component information dictionary to add

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.

See also

e3.anod.store.interface.LocalStore.add_component_from_store()

Parameters:

• from_store – the store to retrieve the component from

• setup – the build setup name

• name – the component name (default: “all”)

• platform – the platform (default: “all”)

• date – the build date (default: None)

• specname – the spec name (default: None)

save(filename: pathlib.Path | str | None = None) → None

Interface method implementation.

See also

e3.anod.store.interface.LocalStore.save()

Parameters:

filename – optional path to save the database to (default: None)

bulk_update_from_store(from_store: e3.anod.store.interface.StoreReadInterface, queries: list[dict[str, Any]]) → list[dict[str, Any]]

Interface method implementation.

See also

e3.anod.store.interface.LocalStore.bulk_update_from_store()

Parameters:

• from_store – the store to retrieve data from

• queries – list of query dictionaries to execute