Bamboostbamboost.index.store
SQLAlchemy Core helper utilities for the BAMBOOST index database.
Attributes
- __all__=
['collections_table', 'create_all', 'collections_upsert_stmt', 'simulations_upsert_stmt', 'parameters_upsert_stmt', 'fetch_collection', 'fetch_collections', 'fetch_collection_uid_by_alias', 'fetch_simulation', 'fetch_simulations', 'fetch_parameters', 'delete_collection_stmt', 'delete_simulation_stmt', 'json_serializer', 'json_deserializer'] - log=
BAMBOOST_LOGGER.getChild(__name__) - _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
(value) -> strConvert a value to a JSON string.
Arguments:
- value:
typing.Any
(value) -> AnyConvert a JSON string to a Python value.
Arguments:
- value:
str
(data) -> InsertCreate 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.
(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.(data) -> InsertArguments:
- data:
typing.Sequence[typing.Mapping[str, typing.Any]] | typing.Mapping[str, typing.Any]
(session, uid) -> CollectionRecord | NoneArguments:
- session:
sqlalchemy.orm.Session - uid:
str
(session) -> list[CollectionRecord]Arguments:
- session:
sqlalchemy.orm.Session
(session, alias) -> str | NoneArguments:
- session:
sqlalchemy.orm.Session - alias:
str
(session, collection_uid, name) -> SimulationRecord | NoneArguments:
- session:
sqlalchemy.orm.Session - collection_uid:
str - name:
str
(session) -> list[SimulationRecord]Arguments:
- session:
sqlalchemy.orm.Session
(session) -> list[ParameterRecord]Arguments:
- session:
sqlalchemy.orm.Session
Arguments:
- uid:
str
(collection_uid, name)Arguments:
- collection_uid:
str - name:
str
(session, row) -> CollectionRecordArguments:
- session:
sqlalchemy.orm.Session - row:
sqlalchemy.RowMapping
(row, parameters) -> SimulationRecordArguments:
- row:
sqlalchemy.RowMapping - parameters:
typing.Sequence[ParameterRecord] | None
(row) -> ParameterRecordArguments:
- row:
sqlalchemy.RowMapping
(session, simulation_ids) -> dict[int, list[ParameterRecord]]Arguments:
- session:
sqlalchemy.orm.Session - simulation_ids:
typing.Sequence[int]
(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.TableThe 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).(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
SqliteJSONEncoder
JSON encoder that handles numpy and datetime types.
(self, obj) -> AnyArguments:
- obj:
typing.Any