Bamboost
bamboostindexbase

bamboost.index.base

Source: https://github.com/smec-ethz/bamboost/tree/main/bamboost/index_/base.py

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

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, return_uid=False)

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=None

    Paths to search for the collection

  • return_uid=False

    If True, return a tuple of (path, uid). The uid is the resolved UID of the collection, which can be different from the input uid if the input uid is an alias.

Index.resolve_uid(self, path) -> CollectionUID

Resolve the UID of a collection from a path.

Arguments:
Returns
CollectionUIDUID of the collection at the path.
Index.sync_collection(self, uid, path=None, *, force_all=False, heal_links=True) -> 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

    If True, sync all simulations regardless of modification time.

  • heal_links:bool=True

    If True, attempt to heal stale links in the collection after sync.

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:
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, SimulationUID]]

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:
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, scan_and_sync=True) -> 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.

  • scan_and_sync:bool=True

    If True, if a target simulation is not found in the index, scan for collections and sync the target collection before trying again. This can be helpful to find targets.

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.heal_stale_links(self, collection_uid=None) -> None

Heal stale links in the database by resolving target_uids in JSON to target_ids.

Arguments:
  • collection_uid:Optional=None

    UID of the collection to heal. If None, heal all collections.

Index._initialize_root_process(self, url) -> None
Arguments:
  • url:str
Index._get_path_and_uid(self, uid) -> tuple[Path | None, str | None]

Fetch the path and UID of a collection given its UID or an alias.

Arguments:
  • uid:str
Index._get_collection_path(self, uid) -> Path | None
Arguments:
  • uid:str
Index._find_collection_uid_by_alias(self, alias) -> str | None
Arguments:
  • alias:str
Index._get_collections(self) -> list[store.CollectionRecord]
Index._get_simulation(self, collection_uid, simulation_name) -> store.SimulationRecord | None
Arguments:

index

Module for indexing BAMBOOST collections.

filtering

Next Page

On this page

Attributes
Functions
_sql_transaction
Classes
LazyDefaultIndex
__delete__
__set__
__get__
Index
sql_transaction
scan_for_collections
check_integrity
resolve_path
resolve_uid
sync_collection
collection
simulation
simulations
collection_links
collection_links_map
insert_linked_sim_parameters
backlinks
drop_collection
drop_simulation
upsert_collection
upsert_simulation
update_simulation_metadata
update_simulation_links
update_simulation_parameters
heal_stale_links
_initialize_root_process
_get_path_and_uid
_get_collection_path
_find_collection_uid_by_alias
_get_collections
_get_simulation