msflib.db

Part of the msflib core package.

msflib.db

cli

The initial_data entry point shared by downstream applications.

An application's initial_data.py only supplies what is its own (the engine, its :class:~msflib.db.migrations.MigrationConfig and a seed function)::

from msflib.db.cli import run_initial_data_cli

def main(argv=None) -> int:
    return run_initial_data_cli(
        engine=engine,
        config=MIGRATION,
        seed=seed_initial_data,
        environment=settings.ENVIRONMENT,
        argv=argv,
    )

if __name__ == "__main__":
    sys.exit(main())

python -m app.initial_data migrates the database (see :func:~msflib.db.migrations.migrate) and seeds it; it is safe to run on every start and never drops data. --reset drops everything first and is refused when environment is production or prod; on any non-local database it also needs --yes or an interactive confirmation (typing RESET; the seed CLI's classifier decides which databases count as non-local). --check only reports what would be done (state, pending revisions, drift) and changes nothing; exit code 0 means there is nothing to do, 3 means changes are pending.

run_initial_data_cli(*, engine: Engine, config: MigrationConfig, seed: Callable[[Engine], None], environment: str | None, argv: Sequence[str] | None = None, description: str = 'Migrate the database and seed the initial data.', reset: Callable[[Engine], object] | None = None, use_alembic: bool | None = None) -> int

Migrate (or with --reset, rebuild) the database and seed it. Returns an exit code.

reset replaces the default reset when the application has extra state to clear, e.g. caches; it must still call :func:~msflib.db.reset.reset_database.

init_db

init_db(engine: Engine, metadata: MetaData, create_tables=False, seeder_config: SeederConfig | None = None) -> None

Initializes the database by optionally creating tables and running seeders. This function should be called after all SQLModel models have been imported in the script to ensure metadata is correctly populated. Table creation is optional and intended for non-Alembic workflows (e.g., quick prototypes or testing). Args: engine (Engine): SQLAlchemy Engine instance connected to the target database. metadata (MetaData): SQLAlchemy MetaData containing table definitions. create_tables (bool, optional): If True, drops all existing tables and creates new ones based on the provided metadata. Defaults to False. seeder_config (SeederConfig, optional): Configuration for running the data seeding process. If provided, it will use the defined seeders and data paths to populate the database. Notes: - This drops every table in metadata when create_tables=True. For deployed apps use msflib.db.migrations.migrate (and msflib.db.cli.run_initial_data_cli), which never drop data; see the core README. - Alembic is recommended for managing schema migrations in production. - All models must be imported before calling this function to register them in metadata.

migrations

Bring a database to the current schema, whatever state it is in.

Deploy scripts should not have to know whether a database is new, tracked by Alembic, or older than Alembic tracking. :func:migrate inspects it and picks:

• empty database: create_all and stamp Alembic at head. A historic migration chain rarely builds a schema from nothing, and does not need to. The catch: create_all only knows the tables in the metadata, so migration-only operations (data inserts, views, triggers, functions) are not applied to a new database. If your chain has such steps, set MigrationConfig.fresh_database="upgrade" to run the whole chain on an empty database instead.
• Alembic-tracked database: alembic upgrade head.
• tables but no version row (kept in sync with create_all so far): stamp baseline_revision, then upgrade head. Migrations after the baseline should tolerate tables and columns that create_all already added (see the README for the pattern).
• SQLite (development and tests): create_all, no Alembic.
• no Alembic revisions yet (an app that has not generated its first migration): create_all, which creates missing tables but never alters existing ones. Nothing is dropped.

Two separate guarantees:

• Atomic (PostgreSQL and SQLite): every path runs in one transaction (on SQLite an explicit one is begun), so a failure leaves the database unchanged. This relies on transactional DDL. On a database whose DDL commits implicitly (MySQL, MariaDB) earlier statements stay when a later one fails, so a failed migration can leave it partly migrated: back up first and rehearse.

A revision that needs a statement PostgreSQL cannot run in a transaction (CREATE INDEX CONCURRENTLY, ALTER TYPE ... ADD VALUE on old servers) therefore cannot run through :func:migrate: PostgreSQL refuses it, or Alembic raises (autocommit_block()), the whole migration rolls back and nothing is committed. Write such a revision with with op.get_context().autocommit_block(): around the statement (a bare statement fails under the standalone command too, which also runs inside a transaction), apply it on its own with alembic upgrade, then run :func:migrate. - Serialized (PostgreSQL only): the transaction holds a Postgres advisory lock, so concurrent starts (several containers or workers) take turns. Other databases are not locked; run one start at a time there.

Alembic is an optional dependency (pip install msflib[migrations] or add alembic to the app). It is only imported when a migration actually runs.

MigrationError

Bases: RuntimeError

The database cannot be migrated as configured (configuration or state problems).

Also raised for Alembic's own command failures. Errors raised by a revision itself, or by the database driver while it runs (sqlalchemy.exc.DBAPIError, anything a revision raises), are not wrapped: they propagate unchanged, after the migration has been rolled back.

MigrationConfig(metadata: MetaData, script_location: str | Path, baseline_revision: str | None = None, fresh_database: Literal['create_all', 'upgrade'] = 'create_all', advisory_lock_id: int = DEFAULT_ADVISORY_LOCK_ID) dataclass

What migrate needs to know about an application.

Attributes:

Name	Type	Description
metadata	MetaData	the application's SQLAlchemy metadata (all models must be imported).
script_location	str | Path	the Alembic script directory (the one with env.py and versions/).
baseline_revision	str | None	revision that databases without version tracking are assumed to match. Required only if such databases exist.
fresh_database	Literal['create_all', 'upgrade']	what to do with an empty database. "create_all" (default) builds the tables from the metadata and stamps head, which skips migration-only operations; "upgrade" runs the whole migration chain instead.

alembic.ini is deliberately not loaded here: its logging setup (fileConfig in a typical env.py) would reconfigure the application's logging on every migration. The ini is still used by standalone alembic commands.

DatabaseReport(state: SchemaState, action: MigrationAction, current_revision: str | None, head_revision: str | None, pending: tuple[str, ...], drift: tuple[str, ...], needs_baseline: bool = False) dataclass

What :func:migrate would do to a database, computed without changing it.

up_to_date: bool property

True when running migrate would change nothing.

alembic_config(config: MigrationConfig, connection: Connection) -> Config

An Alembic Config whose env.py runs on connection (see :func:run_alembic_env).

log_drift(connection: Connection, metadata: MetaData) -> None

Warn about differences between the models and the database (never raises).

has_revisions(config: MigrationConfig) -> bool

Whether the app has any Alembic revision yet.

A script directory that does not exist is an error, not "no revisions": a mistyped path or a directory left out of the deployed package would otherwise fall back to create_all and the deploy would succeed without the migrations. Only an existing Alembic project whose versions/ is empty counts as having none.

inspect_database(engine: Engine, config: MigrationConfig, *, use_alembic: bool | None = None) -> DatabaseReport

Report what :func:migrate would do, without changing anything (safe on production).

migrate(engine: Engine, config: MigrationConfig, *, use_alembic: bool | None = None) -> MigrationAction

Bring the database to the current schema and return the action taken.

use_alembic defaults to True except on SQLite, which is meant for development and tests and is managed with create_all.

run_alembic_env(context: Any, target_metadata: MetaData, *, database_url: Any = None, compare_type: bool = True, **configure_kwargs: Any) -> None

The body of an application's alembic/env.py.

Runs on the connection that :func:migrate provides, so the schema changes, the version stamp and the advisory lock share one transaction. Standalone alembic commands (autogenerate, manual upgrades) still work: they use database_url, or sqlalchemy.url from alembic.ini.

env.py::

from alembic import context
from msflib.db.migrations import run_alembic_env
from sqlmodel import SQLModel
from app import models  # noqa: F401  (registers the tables)
from app.core.config import settings

run_alembic_env(context, SQLModel.metadata, database_url=settings.database_url)

reset

Destructive database reset, with a guard that production databases cannot get around.

ProductionGuardError

Bases: RuntimeError

A destructive operation was attempted with ENVIRONMENT set to production (or prod).

ensure_reset_supported(engine: Engine) -> None

Raise :class:~msflib.db.migrations.MigrationError unless the reset can handle the dialect.

Elsewhere (MySQL, MariaDB, SQL Server, Oracle) foreign keys stay enforced and table names are not listed in dependency order, so a reset could fail half way through. PostgreSQL drops with CASCADE and SQLite runs with enforcement switched off.

reset_database(engine: Engine, config: MigrationConfig, *, environment: str | None, use_alembic: bool | None = None) -> MigrationAction

Drop everything and rebuild the schema with :func:migrate (same use_alembic choice).

Raises :class:ProductionGuardError when environment is production or prod (pass settings.ENVIRONMENT). This is for tests and local development only.

sqlite

SQLite engine setup.

enable_savepoints(engine: Engine) -> Engine

Emit BEGIN eagerly so releasing a first savepoint does not commit on pysqlite.

sqlite_savepoints_enabled(bind: Engine | Connection) -> bool

False only for a SQLite engine not set up with enable_savepoints.

begin_sqlite_transaction(connection: Connection) -> None

Emit the BEGIN pysqlite defers, so releasing a following savepoint cannot commit.