Database Utilities (ixmp4.db.utils)

Submodules

ixmp4.db.utils.alembic module

ixmp4.db.utils.alembic.downgrade_database(dsn: str, revision: str) → None

Uses alembic to downgrade the given database.

ixmp4.db.utils.alembic.get_database_revision(dsn: str) → str | None

Returns the current revision hash of a given database. To do this a connection to the database and a call to alembic are required.

ixmp4.db.utils.alembic.get_head_revision() → str | tuple[str, ...] | None

Returns the revision hash of the newest migration in the alembic versions/ directory.

ixmp4.db.utils.alembic.stamp_database(dsn: str, revision: str) → None

Uses alembic to stamp the given database with the given revision. Only touches the alembic_version table, nothing else.

ixmp4.db.utils.alembic.upgrade_database(dsn: str, revision: str = 'head') → None

Uses alembic to upgrade the given database. Optionally a desired revision can be supplied instead of the default head (newest) revision.

ixmp4.db.utils.sqlite module

ixmp4.db.utils.sqlite.get_database_dir() → Path

Returns the path to the local sqlite database directory.

ixmp4.db.utils.sqlite.get_database_path(name: str) → Path

Returns a Path object for a given sqlite database name. Does not check whether or not the file actually exists.

ixmp4.db.utils.sqlite.get_dsn(database_path: Path) → str

Returns sqlalchemy-friendly sqlite database URI for a given database name.

ixmp4.db.utils.sqlite.search_databases(name: str) → str | None

Returns a database URI if the desired database exists, otherwise None.

Module contents

ixmp4.db.utils.collect_columns_to_select(columns: ColumnCollection[str, ColumnElement[Any]], exclude: Iterable[str]) → ColumnCollection[str, ColumnElement[Any]]

ixmp4.db.utils.create_id_map_subquery(old_exc: Subquery, new_exc: Subquery) → Subquery

ixmp4.db.utils.get_columns(model_class: type[BaseModel]) → ReadOnlyColumnCollection[str, ColumnElement[Any]]

ixmp4.db.utils.get_foreign_columns(model_class: type[BaseModel]) → ReadOnlyColumnCollection[str, ColumnElement[int]]

ixmp4.db.utils.get_pk_columns(model_class: type[BaseModel]) → ReadOnlyColumnCollection[str, ColumnElement[int]]

ixmp4.db.utils.is_joined(exc: Select, model: type[BaseModel]) → bool

Returns True if model has been joined in exc.

ixmp4.db.utils.where_matches_kwargs(exc: Select, model_class: type[BaseModel], **kwargs: Any) → Select