Bamboost
bamboost/Index/sqlmodel

bamboost.index.sqlmodel

Source: https://gitlab.com/cmbm-ethz/bamboost/-/blob/next/bamboost/index/sqlmodel.py

SQLAlchemy ORM models for the bamboost index database.

The SQL model consists of three tables:

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

Simulations are linked to collections via a foreign key, and parameters are linked to simulations via foreign keys.

Attributes

  • log=BAMBOOST_LOGGER.getChild(__name__)
  • _Base=declarative_base()
  • create_all=bamboost.index.sqlmodel._Base.bamboost.index.sqlmodel._Base.metadata.bamboost.index.sqlmodel._Base.metadata.create_all
  • _APIMethod=TypeVar('_APIMethod', bound=Callable[..., Any])
  • _encoders:typing.Dict[type, typing.Callable[[typing.Any], typing.Any]]={datetime.datetime: lambda obj: obj.obj.isoformat(), complex: lambda obj: {'real': obj.obj.real, 'imag': obj.obj.imag}}
  • _decoders:typing.Dict[type, typing.Callable[[typing.Any], typing.Any]]={datetime.datetime: lambda obj: datetime.datetime.datetime.datetime.fromisoformat(obj), complex: lambda obj: complex(obj['real'], obj['imag'])}

Functions

json_serializer(value) -> str

Convert a value to a JSON string.

Arguments:
  • value:typing.Any

    value to convert

Returns
strJSON string
json_deserializer(value) -> Any

Convert a JSON string to a value.

Arguments:
  • value:str

    JSON string to convert

Returns
typing.AnyConverted value

Classes

SqliteJSONEncoder

Custom JSON encoder for numpy types.

SqliteJSONEncoder.default(self, obj) -> Any
Arguments:
  • obj:typing.Any

_CollectionMixin

Methods for collections.

Attributes:
  • simulations:sqlalchemy.orm.Mapped[typing.List[SimulationORM]]
  • parameters:typing.List[ParameterORM]

    Retrieves all parameters associated with the collections's simulations.

_CollectionMixin.get_parameter_keys(self) -> tuple[list[str], list[int]]

Extracts unique parameter keys and their occurences across all simulations.

Returns
tuple[list[str], list[int]]A tuple containing a list of unique parameter keys and a corresponding count list.
_CollectionMixin.to_pandas(self, flatten=True) -> 'DataFrame'

Converts the collection to a pandas DataFrame.

Arguments:
  • flatten:bool=True

    If True, flatten dictionaries with dot notation.

Returns
'DataFrame'pandas.DataFrame: DataFrame representation of the collection.

CollectionORM

ORM model representing a collection of simulations.

Attributes:
  • __tablename__=bamboost.constants.TABLENAME_COLLECTIONS
  • uid:sqlalchemy.orm.Mapped[str]=mapped_column(primary_key=True)

    Unique identifier for the collection (primary key)

  • path:sqlalchemy.orm.Mapped[str]=mapped_column(String)

    file system path where the collection is stored.

  • simulations:sqlalchemy.orm.Mapped[typing.List[SimulationORM]]=relationship('SimulationORM', back_populates='collection', cascade='all, delete-orphan')

    Relationship to the associated simulations, with cascade delete-orphan.

CollectionORM.__repr__(self) -> str
CollectionORM.upsert(cls, data) -> Insert

Inserts or updates collection records based on the unique uid.

Arguments:
  • cls
  • data:typing.Sequence[typing.Dict[str, str]] | typing.Dict[str, str]

    Data to be upserted. Can be a single dictionary or a sequence of dictionaries.

Returns
sqlalchemy.InsertSQLAlchemy insert statement with conflict resolution.
CollectionORM._repr_html_(self) -> str

FilteredCollection

FilteredCollection(self, base, filter)

In-memory filtered view of a CollectionORM.

Arguments:
Attributes:
  • _base=bamboost.index.sqlmodel.FilteredCollection(base)
  • _filter=bamboost.index.sqlmodel.FilteredCollection(filter)
  • simulations:list[SimulationORM]

    List of simulations in the collection that match the filter criteria.

  • uid:str
  • path:str
FilteredCollection.__repr__(self) -> str
FilteredCollection.to_pandas(self) -> 'DataFrame'

Converts the collection to a pandas DataFrame.

Returns
'DataFrame'pandas.DataFrame: DataFrame representation of the collection.
FilteredCollection._repr_html_(self) -> str

SimulationORM

ORM model representing a simulation in a collection.

Attributes:
  • __tablename__=bamboost.constants.TABLENAME_SIMULATIONS
  • __table_args__=(sqlalchemy.UniqueConstraint('collection_uid', 'name', name='uix_collection_name'),)
  • id:sqlalchemy.orm.Mapped[int]=mapped_column(primary_key=True, autoincrement=True, unique=True)

    Unique simulation ID (primary key)

  • collection_uid:sqlalchemy.orm.Mapped[str]=mapped_column(ForeignKey(CollectionORM.uid))

    Foreign key linking to CollectionORM

  • name:sqlalchemy.orm.Mapped[str]=mapped_column(String, nullable=False)

    Name of the simulation

  • created_at:sqlalchemy.orm.Mapped[sqlalchemy.DateTime]=mapped_column(DateTime, nullable=False, default=datetime.now)

    Timestamp when the simulation was created

  • modified_at:sqlalchemy.orm.Mapped[sqlalchemy.DateTime]=mapped_column(DateTime, nullable=False, default=datetime.now)

    Timestamp when the simulation was last modified

  • description:sqlalchemy.orm.Mapped[typing.Optional[str]]=mapped_column(String)

    Optional description of the simulation

  • status:sqlalchemy.orm.Mapped[str]=mapped_column(String, nullable=False, default='initialized')

    Current status of the simulation

  • submitted:sqlalchemy.orm.Mapped[bool]=mapped_column(Boolean, nullable=False, default=False)

    Indicates whether the simulation has been submitted

  • collection:sqlalchemy.orm.Mapped[CollectionORM]=relationship('CollectionORM', back_populates='simulations')

    Relationship to the parent collection

  • parameters:sqlalchemy.orm.Mapped[typing.List[ParameterORM]]=relationship('ParameterORM', back_populates='simulation', cascade='all, delete-orphan')

    List of associated parameters

  • parameter_dict:typing.Dict[str, typing.Any]

    Return a dictionary of parameters associated with the simulation.

SimulationORM.__repr__(self) -> str
SimulationORM.upsert(cls, data) -> ReturningInsert[Tuple[int]]

Generate an upsert (insert or update) statement.

Arguments:
  • cls
  • data:typing.Sequence[dict] | dict

    Data to upsert, either a single record or a sequence of records.

Returns
sqlalchemy.sql.dml.ReturningInsert[typing.Tuple[int]]SQLAlchemy insert statement with conflict resolution. Returning the simulation ID.
SimulationORM.as_dict_metadata(self) -> Dict[str, Any]
SimulationORM.as_dict(self, standalone=True) -> Dict[str, Any]

Return the simulation as a dictionary.

Arguments:
  • standalone:bool=True

    If False, "id", "collection_uid", and "modified_at" are excluded. Defaults to True.

Returns
typing.Dict[str, typing.Any]Dictionary representation of the simulation.

ParameterORM

ORM model representing a parameter associated with a simulation.

Attributes:
  • __tablename__=bamboost.constants.TABLENAME_PARAMETERS
  • __table_args__=(sqlalchemy.UniqueConstraint('simulation_id', 'key', name='uix_simulation_key'),)
  • id:sqlalchemy.orm.Mapped[int]=mapped_column(primary_key=True, autoincrement=True)

    Unique parameter ID (primary key)

  • simulation_id:sqlalchemy.orm.Mapped[int]=mapped_column(ForeignKey(SimulationORM.id, ondelete='CASCADE'), nullable=False)

    Foreign key linking to SimulationORM

  • key:sqlalchemy.orm.Mapped[str]=mapped_column(String, nullable=False)

    Parameter key (name)

  • value:sqlalchemy.orm.Mapped[sqlalchemy.JSON]=mapped_column(JSON, nullable=False)

    Parameter value (stored as JSON)

  • simulation:sqlalchemy.orm.Mapped[SimulationORM]=relationship('SimulationORM', back_populates='parameters')

    Relationship to the parent simulation

ParameterORM.__repr__(self) -> str
ParameterORM.upsert(cls, data) -> Insert

Generate an upsert (insert or update) statement.

Arguments:
  • cls
  • data:typing.Sequence[ParameterORM._dataT] | bamboost.index.sqlmodel.ParameterORM._dataT

    Data to upsert, either a single record or a sequence of records.

Returns
sqlalchemy.InsertInsert statement with conflict resolution.

base

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

mpi

This module handles the detection and selection of the appropriate MPI implementation for Bamboost, either using `mpi4py` for real MPI environments or falling back to a mock MPI implementation for non-MPI environments.

On this page

Attributes
Functions
json_serializer
json_deserializer
Classes
SqliteJSONEncoder
default
_CollectionMixin
get_parameter_keys
to_pandas
CollectionORM
__repr__
upsert
_repr_html_
FilteredCollection
__repr__
to_pandas
_repr_html_
SimulationORM
__repr__
upsert
as_dict_metadata
as_dict
ParameterORM
__repr__
upsert