SQLAlchemy Core schema definitions and helper utilities for the BAMBOOST index database.

Attributes

log=BAMBOOST_LOGGER.getChild(__name__)
metadata=MetaData()
collections_table=Table(TABLENAME_COLLECTIONS, metadata, Column('uid', String, primary_key=True), Column('path', String, nullable=False), Column('created_at', DateTime, nullable=True), Column('description', String, nullable=False, default='', server_default=''), Column('tags', JSON, nullable=False, default=list, server_default='[]'), Column('aliases', JSON, nullable=False, default=list, server_default='[]'), Column('author', JSON, nullable=True, default=None, server_default='null'))
simulations_table=Table(TABLENAME_SIMULATIONS, metadata, Column('id', Integer, primary_key=True, autoincrement=True), Column('collection_uid', String, ForeignKey(f'{TABLENAME_COLLECTIONS}.uid', ondelete='CASCADE')), Column('name', String, nullable=False), Column('created_at', DateTime, nullable=False, default=(datetime.now)), Column('modified_at', DateTime, nullable=False, default=(datetime.now)), Column('description', String, nullable=True), Column('tags', JSON, nullable=False, default=list, server_default='[]'), Column('status', String, nullable=False, default='initialized', server_default='initialized'), Column('submitted', Boolean, nullable=False, default=False, server_default='0'), UniqueConstraint('collection_uid', 'name', name='uix_collection_name'))
simulation_links_table=Table(TABLENAME_SIMULATION_LINKS, metadata, Column('id', Integer, primary_key=True, autoincrement=True), Column('source_id', Integer, ForeignKey(f'{TABLENAME_SIMULATIONS}.id', ondelete='CASCADE'), nullable=False), Column('target_id', Integer, ForeignKey(f'{TABLENAME_SIMULATIONS}.id', ondelete='CASCADE'), nullable=False), Column('name', String, nullable=False), UniqueConstraint('source_id', 'name', name='uix_source_link_name'))
parameters_table=Table(TABLENAME_PARAMETERS, metadata, Column('id', Integer, primary_key=True, autoincrement=True), Column('simulation_id', Integer, ForeignKey(f'{TABLENAME_SIMULATIONS}.id', ondelete='CASCADE'), nullable=False), Column('key', String, nullable=False), Column('value', JSON, nullable=False), UniqueConstraint('simulation_id', 'key', name='uix_simulation_key'))
_ENCODERS: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:dict[type, typing.Callable[[typing.Any], typing.Any]]={datetime.datetime: datetime.datetime.datetime.datetime.fromisoformat, complex: lambda payload: complex(payload['real'], payload['imag'])}

Functions

create_all(engine) -> None

Create all tables defined in this module.

Arguments:

engine:sqlalchemy.Engine

json_serializer(value) -> str

Convert a value to a JSON string.

Arguments:

value:typing.Any

json_deserializer(value) -> Any

Convert a JSON string to a Python value.

Arguments:

value:str

collections_upsert_stmt(data) -> Insert

Create an upsert statement for collections.

Arguments:

data:typing.Sequence[typing.Mapping[str, typing.Any]] | typing.Mapping[str, typing.Any]
A single dictionary or a sequence of dictionaries representing the collections to be inserted or updated.

simulations_upsert_stmt(data) -> ReturningInsert[Any]

Create an upsert statement for simulations.

Arguments:

data:typing.Sequence[typing.Mapping[str, typing.Any]] | typing.Mapping[str, typing.Any]
A single dictionary or a sequence of dictionaries representing the simulations to be inserted or updated.

Returns

sqlalchemy.sql.dml.ReturningInsert[typing.Any]A SQLAlchemy Insert statement with a RETURNING clause for the simulation IDs.

parameters_upsert_stmt(data) -> Insert

Arguments:

data:typing.Sequence[typing.Mapping[str, typing.Any]] | typing.Mapping[str, typing.Any]

simulation_links_upsert_stmt(data) -> Insert

Arguments:

data:typing.Sequence[typing.Mapping[str, typing.Any]] | typing.Mapping[str, typing.Any]

delete_collection_stmt(uid) -> Delete

Arguments:

uid:str

delete_simulation_stmt(collection_uid, name) -> Delete

Arguments:

collection_uid:str
name:str

delete_simulation_links_stmt(source_id) -> Delete

Arguments:

source_id:int

get_collection_uids(session) -> tuple[str, ...]

Fetch all collection UIDs without loading simulations.

Arguments:

session:sqlalchemy.orm.Session

fetch_collection(session, uid) -> CollectionRecord | None

Arguments:

session:sqlalchemy.orm.Session
uid:str

fetch_collections(session) -> list[CollectionRecord]

Arguments:

session:sqlalchemy.orm.Session

fetch_collection_uid_by_alias(session, alias) -> str | None

Arguments:

session:sqlalchemy.orm.Session
alias:str

fetch_simulation_id(session, collection_uid, simulation_name) -> None | int

Fetch the ID of a simulation given its collection UID and name.

Arguments:

session:sqlalchemy.orm.Session
collection_uid:str
simulation_name:str

fetch_simulation_uid(session, simulation_id) -> SimulationUID | None

Fetch the UID of a simulation given its ID.

Arguments:

session:sqlalchemy.orm.Session
simulation_id:int

fetch_simulation(session, collection_uid, name) -> SimulationRecord | None

Fetch a simulation given its collection UID and name.

Arguments:

session:sqlalchemy.orm.Session
collection_uid:str
name:str

fetch_simulations_by_uid(session, uids) -> list[SimulationRecord]

Fetch multiple simulations by their collection UID and name.

Arguments:

session:sqlalchemy.orm.Session
uids:typing.Sequence[tuple[str, str]]

fetch_simulations(session, collection_uids=None) -> list[SimulationRecord]

Fetch all simulations in the database.

Arguments:

session:sqlalchemy.orm.Session
collection_uids:list[str] | None=None

fetch_parameters(session) -> list[ParameterRecord]

Arguments:

session:sqlalchemy.orm.Session

fetch_links(session) -> list[LinkRecord]

Arguments:

session:sqlalchemy.orm.Session

fetch_collection_links(session, collection_uid) -> list[LinkRecord]

Fetch all links for a collection.

Arguments:

session:sqlalchemy.orm.Session
collection_uid:str

fetch_collection_links_map(session, collection_uid) -> dict[str, dict[str, str]]

Fetch all links for a collection as a mapping from simulation name to links.

Arguments:

session:sqlalchemy.orm.Session
collection_uid:str

fetch_backlinks(session, collection_uid, simulation_name) -> Sequence[RowMapping]

Fetch all simulations that link to the given simulation.

Arguments:

session:sqlalchemy.orm.Session
collection_uid:str
simulation_name:str

_column_sql_for_alter(engine, column) -> str

Build a SQLite-safe column definition for ALTER TABLE ADD COLUMN.

Arguments:

engine:sqlalchemy.Engine
column:sqlalchemy.Column[typing.Any]

_server_default_sql(column) -> str | None

Arguments:

column:sqlalchemy.Column[typing.Any]

_fetch_parameters_for(session, simulation_ids) -> dict[int, list[ParameterRecord]]

Arguments:

session:sqlalchemy.orm.Session
simulation_ids:typing.Iterable[int]

_fetch_links_for(session, simulation_ids) -> dict[int, dict[str, SimulationUID]]

Arguments:

session:sqlalchemy.orm.Session
simulation_ids:typing.Sequence[int]

_normalize_payload(data, table) -> tuple[Sequence[dict[str, Any]] | dict[str, Any], bool]

Ensure the payload is a list of dictionaries with only valid keys for the table.

Arguments:

data:typing.Sequence[typing.Mapping[str, typing.Any]] | typing.Mapping[str, typing.Any]
A single dictionary or a sequence of dictionaries representing the records to be inserted or updated.
table:sqlalchemy.Table
The SQLAlchemy Table object representing the target database table.

Returns

tuple[typing.Sequence[dict[str, typing.Any]] | dict[str, typing.Any], bool]A tuple containing: - A list of dictionaries (or a single dictionary if the input was a single record) with keys filtered to match the table's columns. - A boolean indicating whether the input was a single record (True) or multiple records (False).

_collect_update_columns(payload, excluded_keys) -> set[str]

Collect the set of columns to be updated in an upsert operation.

Arguments:

payload:typing.Sequence[dict[str, typing.Any]] | dict[str, typing.Any]
A single dictionary or a sequence of dictionaries representing the records to be inserted or updated.
excluded_keys:set[str]
A set of keys to exclude from the update operation (e.g., primary keys).

Classes

ParameterRecord

ParameterRecord(self, id, simulation_id, key, value) -> None

Arguments:

id:int
simulation_id:int
key:str
value:typing.Any

Attributes:

id:int
simulation_id:int
key:str
value:typing.Any

ParameterRecord.from_mapping(cls, row) -> ParameterRecord

Arguments:

cls
row:sqlalchemy.RowMapping

LinkRecord

LinkRecord(self, id, source_id, target_id, source_uid, name) -> None

Arguments:

id:int
source_id:int
target_id:int
source_uid:SimulationUID
name:str

Attributes:

id:int
source_id:int
target_id:int
source_uid:SimulationUID
name:str

LinkRecord.from_mapping(cls, row) -> LinkRecord

Arguments:

cls
row:sqlalchemy.RowMapping

SimulationRecord

(self, id, collection_uid, name, created_at, modified_at, description, status, submitted, tags=list(), parameters=list(), links=dict()) -> None

Arguments:

id:int
collection_uid:str
name:str
created_at:datetime.datetime
modified_at:datetime.datetime
description:str | None
status:str
submitted:bool
tags:list[str]=list()
parameters:list[ParameterRecord]=list()
links:dict[str, SimulationUID]=dict()

Attributes:

id:int
collection_uid:str
name:str
created_at:datetime.datetime
modified_at:datetime.datetime
description:str | None
status:str
submitted:bool
tags:list[str]=field(default_factory=list)
parameters:list[ParameterRecord]=field(default_factory=list)
links:dict[str, SimulationUID]=field(default_factory=dict)
parameter_dict:dict[str, typing.Any]

SimulationRecord.from_mapping(cls, row, parameters, links) -> SimulationRecord

Arguments:

cls
row:sqlalchemy.RowMapping
parameters:list[ParameterRecord]
links:dict[str, SimulationUID]

SimulationRecord.as_dict(self, standalone=True, include_parameters=True, include_links=True) -> dict[str, Any]

Arguments:

standalone:bool=True
include_parameters:bool=True
include_links:bool=True

CollectionMetadata

CollectionMetadata(self, uid, created_at=None, tags=list(), aliases=list(), author=None, description=None) -> None

Collection metadata container. Can load/save from/to dicts, storing unknown fields in an extras dictionary.

Arguments:

uid:str
created_at:datetime.datetime | None=None
tags:list[str]=list()
aliases:list[str]=list()
author:str | dict | None=None
description:str | None=None

Attributes:

uid:str
Unique identifier of the collection.
created_at:datetime.datetime | None=field(default=None)
Creation timestamp.
tags:list[str]=field(default_factory=list)
List of tags associated with the collection.
aliases:list[str]=field(default_factory=list)
List of aliases for the collection.
author:str | dict | None=field(default=None)
Author information, can be a string or a dictionary.
description:str | None=field(default=None)
Completely optional description of the collection.
_extras:dict[str, typing.Any]=field(default_factory=dict, init=False, repr=False)

CollectionMetadata.__post_init__(self)

CollectionMetadata.to_dict(self) -> dict[str, Any]

CollectionMetadata.from_dict(cls, data, *_) -> Self

Create an instance from a dictionary, storing unknown fields in the extras dictionary.

Arguments:

cls
data:dict[str, typing.Any]
The input dictionary.
_=()

CollectionRecord

(self, uid, created_at=None, tags=list(), aliases=list(), author=None, description=None, path='', simulations=list()) -> None

Collection record, including metadata and associated simulations. Used for exposing (read only) a record in the collection table in the sql database.

Arguments:

uid:str
created_at:datetime.datetime | None=None
tags:list[str]=list()
aliases:list[str]=list()
author:str | dict | None=None
description:str | None=None
path:str=''
simulations:list[SimulationRecord] | property=list()

Attributes:

path:str=field(default='')
Path to the collection on disk. Guaranteed to be not null.
simulations:list[SimulationRecord] | property=field(default_factory=list, repr=False)
List of simulations associated with the collection.
parameters:list[ParameterRecord]
links:dict[str, dict[str, SimulationUID]]
Dictionary of links for each simulation in the collection.

Bases

CollectionMetadata10

CollectionRecord.from_mapping(cls, row, simulations) -> CollectionRecord

Arguments:

cls
row:sqlalchemy.RowMapping
simulations:list[SimulationRecord]

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

CollectionRecord.to_pandas(self, flatten=True) -> 'DataFrame'

Arguments:

flatten:bool=True

CollectionRecord.filtered(self, filter, sorter) -> Self

Arguments:

filter:Filter | None
sorter:Sorter | None

SqliteJSONEncoder

JSON encoder that handles numpy and datetime types.

SqliteJSONEncoder.default(self, obj) -> Any

Arguments:

obj:typing.Any

bamboost.index.store

Attributes

Functions

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Returns

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Returns

Arguments:

Classes

ParameterRecord

Arguments:

Attributes:

Arguments:

LinkRecord

Arguments:

Attributes:

Arguments:

SimulationRecord

Arguments:

Attributes:

Arguments:

Arguments:

CollectionMetadata

Arguments:

Attributes:

Arguments:

CollectionRecord

Arguments:

Attributes:

Bases

Arguments:

Arguments:

Arguments:

SqliteJSONEncoder

Arguments:

On this page