Indexing of bamboost collections and their simulations/parameters. SQLAlchemy is used to interact with the SQLite database.

The index is generated on the fly or can be explicitly created by scanning the search_paths for collections. The index is stored as a SQLite database that stores the path of collections (characterized with a unique UID), as well as the metadata and parameters of all simulations.

The Index class provides the public API for interacting with the index. This works in paralell execution, but the class is designed to execute any operations on the database on the root process only. Methods that return something use bcast to cast the result to all processes. Any SQL operation is executed only on the root process!

Database schema:

collections: Contains information about the collections, namely uids and corresponding paths.
simulations: Contains information about the simulations, including names, statuses, and links to the corresponding parameters.
parameters: Contains the parameters associated with the simulations.

Attributes

log=BAMBOOST_LOGGER.getChild('Database')
P=ParamSpec('P')
T=TypeVar('T')

Functions

_sql_transaction(func) -> Callable[Concatenate[Index, P], T]

Decorator to add a session to the function signature.

Arguments:

func:typing.Callable[typing_extensions.Concatenate[Index, bamboost.index.base.P], bamboost.index.base.T]
The function to decorate.

Classes

LazyDefaultIndex

LazyDefaultIndex(self) -> None

Attributes:

_instance=None

LazyDefaultIndex.__delete__(self, instance) -> None

Arguments:

instance:None

LazyDefaultIndex.__set__(self, instance, value) -> None

Arguments:

instance:None
value:Index

LazyDefaultIndex.__get__(self, instance, owner) -> Index

Arguments:

instance:None
owner:type[Index]

Index

Index(self, sql_file=None, comm=None, *, search_paths=None) -> None

API for indexing BAMBOOST collections and simulations.

Arguments:

sql_file:StrPath | None=None
comm:Comm | None=None
search_paths:typing.Iterable[str | pathlib.Path] | None=None

Attributes:

_comm=Communicator()
_engine:sqlalchemy.Engine
_sm:typing.Callable[..., sqlalchemy.orm.Session]
_s:sqlalchemy.orm.Session
search_paths:PathSet=PathSet(search_paths or config.index.searchPaths)
Paths to scan for collections.
default:LazyDefaultIndex=LazyDefaultIndex()
A default index instance. Uses the default SQLite database file and search paths from the configuration.
_file=bamboost.index.base.Index(sql_file) or bamboost._config.config.bamboost._config.config.index.bamboost._config.config.index.databaseFile
The path to the SQLite database file.
_isolated=bamboost._config.config.bamboost._config.config.index.bamboost._config.config.index.isolated
Whether project based indexing is used.
_url=f'sqlite:///{bamboost.index.base.Index(self).bamboost.index.base.Index(self)._file}'
The URL to the SQLite database file.
all_collections:list[bamboost.index.store.CollectionRecord]
Return all collections in the index.
all_simulations:list[bamboost.index.store.SimulationRecord]
Return all simulations in the index.
all_parameters:list[bamboost.index.store.ParameterRecord]
Return all parameters in the index.
all_links:list[bamboost.index.store.LinkRecord]
Return all simulation links in the index.

Usage

Create an instance of the Index class and use its methods to interact with the index. $ from bamboost.index import Index $ index = Index()

Scan for collections in known paths: $ index.scan_for_collections()

Resolve the path of a collection: $ index.resolve_path()

Get a simulation from its collection and simulation name: $ index.get_simulation(, )

Index.sql_transaction(self) -> Generator[Session, None, None]

Context manager for a SQL transaction.

If no transaction is active, a new transaction is started. If a transaction is active, the current session is used.

Usage

>>> with index.sql_transaction() as s:
...     s.execute(...)

Index.scan_for_collections(self, *, search_paths=None) -> list[tuple[str, Path]]

Scan known paths for collections and update the index.

Iterates through the search paths and searches files with the identifier file structure. If a collection is found, it is added to the cache.

Arguments:

search_paths:List[pathlib.Path]=None
Paths to scan for collections. Defaults to config.index.searchPaths.

Index.check_integrity(self) -> None

Check the integrity of the cache.

This method checks if the paths stored in the cache are valid. If a path is not valid, it is removed from the cache.

Index.resolve_path(self, uid, *, search_paths=None) -> Path

Resolve and return the path of a collection from its UID. Raises a FileNotFoundError if the collection is not found in the search paths.

Arguments:

uid:str
UID of the collection
search_paths:set[StrPath] | None=None
Paths to search for the collection

Index.resolve_uid(self, path) -> CollectionUID

Resolve the UID of a collection from a path.

Returns the UID of the collection or a new UID if it can't be determined.

Arguments:

path:StrPath
Path of the collection

Index.sync_collection(self, uid, path=None, *, force_all=False) -> None

Sync the table with the file system.

Iterates through the simulations in the collection and updates the metadata and parameters if the HDF5 file has been modified.

Arguments:

uid:str
UID of the collection
path:Optional=None
Path of the collection
force_all:bool=False

Index.collection(self, uid) -> store.CollectionRecord | None

Return a collection from the index.

Arguments:

uid:str
UID of the collection

Index.simulation(self, uid) -> store.SimulationRecord | None

Return a simulation from the index.

Arguments:

uid:str | SimulationUID

Index.simulations(self, uids) -> list[store.SimulationRecord]

Return multiple simulations from the index.

Arguments:

uids:typing.Iterable[str | SimulationUID]
Iterable of simulation UIDs

Index.collection_links(self, uid) -> list[store.LinkRecord]

Return all simulation links for a specific collection.

Arguments:

uid:str
UID of the collection

Index.collection_links_map(self, uid) -> dict[str, dict[str, str]]

Return a mapping of simulation names to their links for a specific collection.

Arguments:

uid:str
UID of the collection

Index.insert_linked_sim_parameters(self, collection_record, include_links=True) -> store.CollectionRecord

Insert parameters of linked simulations into the simulations of a collection record.

Modifies the collection record in place and returns it.

Arguments:

collection_record:bamboost.index.store.CollectionRecord
The collection record to modify
include_links:typing.Iterable[str] | typing.Literal[True]=True
If provided, only include parameters of linked simulations with link names in this iterable. If True, include parameters of all linked simulations.

Index.backlinks(self, uid) -> list[tuple[SimulationUID, str]]

Return all simulations that link to the given simulation.

Arguments:

uid:str | SimulationUID
UID of the target simulation.

Index.drop_collection(self, uid) -> None

Drop a collection from the cache.

Arguments:

uid:str
UID of the collection

Index.drop_simulation(self, collection_uid, simulation_name) -> None

Drop a simulation from the cache.

Arguments:

collection_uid:str
UID of the collection
simulation_name:str
Name of the simulation

Index.upsert_collection(self, uid, path, metadata=None) -> None

Cache a collection in the index.

Arguments:

uid:str
UID of the collection
path:pathlib.Path
Path of the collection
metadata:typing.Mapping[str, typing.Any] | None=None

Index.upsert_simulation(self, collection_uid, simulation_name, parameters=None, metadata=None, links=None, *, collection_path=None) -> None

Cache a simulation from a collection.

Arguments:

collection_uid:str
UID of the collection
simulation_name:str
Name of the simulation
parameters:typing.Mapping[typing.Any, typing.Any] | None=None
metadata:typing.Mapping[typing.Any, typing.Any] | None=None
links:typing.Mapping[typing.Any, typing.Any] | None=None
collection_path:Optional=None
Path of the collection

Index.update_simulation_metadata(self, collection_uid, simulation_name, data) -> None

Update the metadata of a simulation by passing it as a dict.

Arguments:

collection_uid:str
simulation_name:str
data:typing.Mapping
Dictionary with new data

Index.update_simulation_links(self, collection_uid, simulation_name, links, *, raise_on_invalid_target=config.index.strictLinksWhenSyncing) -> None

Update the links of a simulation.

Arguments:

collection_uid:str
UID of the collection
simulation_name:str
Name of the simulation
links:typing.Mapping[str, typing.Any]
Dictionary with new links
raise_on_invalid_target:bool=bamboost._config.config.bamboost._config.config.index.bamboost._config.config.index.strictLinksWhenSyncing
If True, raise an error if any of the target simulations do not exist in the index. If False, log a warning and skip the invalid links.

Index.update_simulation_parameters(self, collection_uid, simulation_name, parameters) -> None

Update the parameters of a simulation by passing it as a dict.

Arguments:

collection_uid:str
simulation_name:str
parameters:typing.Mapping[str, typing.Any]
Dictionary with new parameters

Index._initialize_root_process(self, url) -> None

Arguments:

url:str

Index._find_collection_uid_by_alias(self, alias) -> str | None

Arguments:

alias:str

Index._get_collection_path(self, uid) -> Path | None

Arguments:

uid:str

Index._get_collections(self) -> list[store.CollectionRecord]

Index._get_simulation(self, collection_uid, simulation_name) -> store.SimulationRecord | None

Arguments:

collection_uid:CollectionUID | str
simulation_name:str

bamboost.index.base

Attributes

Functions

Arguments:

Classes

LazyDefaultIndex

Attributes:

Arguments:

Arguments:

Arguments:

Index

Arguments:

Attributes:

Usage

Usage

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

On this page