simdb.database.database module¶

class simdb.database.database.Database(db_type: DBMS, scopefunc=None, **kwargs)[source]¶

Bases: object

Class to wrap the database access.

class DBMS(value)[source]¶

Bases: Enum

DBMSs supported.

MSSQL = 3¶

POSTGRESQL = 2¶

SQLITE = 1¶

add_watcher(sim_ref: str, watcher: Watcher)[source]¶

close()[source]¶

Close the database session and dispose of the engine.

delete_simulation(sim_ref: str) → Simulation[source]¶

Delete the specified simulation from the database.

Parameters:¶

sim_ref: str¶

The simulation UUID or alias.

Returns:¶

None

engine : Engine¶

get_aliases(prefix: str | None) → list[str][source]¶

get_file(file_uuid_str: str) → File[source]¶

Get the specified file from the database.

Parameters:¶

file_uuid_str: str¶

The string representation of the file UUID.

Returns:¶

The File.

get_metadata(sim_ref: str, name: str) → list[str][source]¶

Get all the metadata for the given simulation with the given key.

Parameters:¶

sim_ref: str¶

the simulation identifier

name: str¶

the metadata key

Returns:¶

The matching metadata values.

get_simulation(sim_ref: str) → Simulation[source]¶

Get the specified simulation from the database.

Parameters:¶

sim_ref: str¶

The simulation UUID or alias.

Returns:¶

The Simulation.

get_simulation_children(simulation: Simulation) → list[dict][source]¶

get_simulation_children_ref(simulation: Simulation) → list[SimulationReference][source]¶

get_simulation_data(query)[source]¶

get_simulation_parents(simulation: Simulation) → list[dict][source]¶

get_simulation_parents_ref(simulation: Simulation) → list[SimulationReference][source]¶

insert_simulation(simulation: Simulation) → None[source]¶

Insert the given simulation into the database.

Parameters:¶

simulation: Simulation¶

The Simulation to insert.

Returns:¶

None

list_files() → list[File][source]¶

Return a list of all the files stored in the database.

Returns:¶

A list of Files.

list_metadata_keys() → list[dict][source]¶

list_metadata_values(name: str) → list[str][source]¶

list_simulation_data(meta_keys: list[str] | None = None, limit: int = 0, page: int = 1, sort_by: str = '', sort_asc: bool = False) → tuple[int, list[dict]][source]¶

Return a list of all the simulations stored in the database.

Returns:¶

A tuple of (total_count, list of simulation data dicts).

list_simulations(meta_keys: list[str] | None = None, limit: int = 0) → list[Simulation][source]¶

Return a list of all the simulations stored in the database.

Returns:¶

A list of Simulations.

list_watchers(sim_ref: str) → list[Watcher][source]¶

query_meta(constraints: list[tuple[str, str, QueryType]]) → list[Simulation][source]¶

Query the metadata and return matching simulations.

Returns:¶

query_meta_data(constraints: list[tuple[str, str, QueryType]], meta_keys: list[str], limit: int = 0, page: int = 1, sort_by: str = '', sort_asc: bool = False) → tuple[int, list[dict]][source]¶

Query the metadata and return matching simulations.

Returns:¶

remove()[source]¶

Remove the current session

remove_watcher(sim_ref: str, username: str)[source]¶

reset() → None[source]¶

Clear all the data out of the database.

Returns:¶

None

session : Session¶

exception simdb.database.database.DatabaseError[source]¶

Bases: RuntimeError

exception simdb.database.database.DatabaseOutdatedError[source]¶

Bases: DatabaseError

exception simdb.database.database.DatabaseUninitializedError[source]¶

Bases: DatabaseError

class simdb.database.database.Session(session_factory, scopefunc=None)[source]¶

Bases: scoped_session

add(obj: Base, *args, **kwargs)[source]¶

commit()[source]¶

delete(obj: Base)[source]¶

execute(query: Any, *args, **kwargs) → Any[source]¶

query(obj: Base, *args, **kwargs) → Any[source]¶

rollback()[source]¶

simdb.database.database.backup_local_db(config: Config)[source]¶

simdb.database.database.check_migrations(engine) → str[source]¶

Check that the database is up-to-date with the latest Alembic migration.

Raises DatabaseUninitializedError if the database has not been initialised at all (i.e. the alembic_version table is absent), or DatabaseOutdatedError if the database schema is behind the head revision.

simdb.database.database.get_local_db(config: Config) → Database[source]¶

simdb.database.database.run_migrations(engine) → None[source]¶

Run the database migrations.