loom.core.repository.sqlalchemy.migrations

Alembic support for row-scoped schemas: the hook, the environment and the runners.

ENV_TEMPLATE_PATH points at the environment a product copies as env.py into each tree; it is not importable outside an Alembic run.

loom.core.repository.sqlalchemy.migrations.alembic_config(script_location, url)

Build a programmatic Alembic configuration for one tree.

A script location whose last path element is data is the data tree; anything else is the structural tree. The URL may be postgresql+asyncpg.

Parameters:

• script_location (str)

• url (str)

Return type:

alembic.config.Config

loom.core.repository.sqlalchemy.migrations.check(config, application)

Fail when the models, the database or the guard registry disagree.

Tables and columns are compared by Alembic’s autogenerate; the registry is compared against the application’s scoped tables. Partitions are in neither: include_object_for() keeps autogenerate away from tables the models do not declare, and a registered partition of a partitioned scoped table is not registry drift. Policy and privilege drift belong to verify. Call it from a thread without a running loop.

Raises:

ConfigError – Naming the first drift found.

Parameters:

• config (Config)

• application (Application)

Return type:

None

loom.core.repository.sqlalchemy.migrations.run_data_migrations(connection, application, *, lock_timeout='5s', statement_timeout='60s')

Run the data tree as a declared bypass user; the structural tree must be at head.

No closing assertion: the bypass user has no access to the guard and no CREATE on the schema, so a data revision cannot change the schema.

Parameters:

• connection (Connection)

• application (Application)

• lock_timeout (str)

• statement_timeout (str)

Return type:

None

loom.core.repository.sqlalchemy.migrations.run_migrations(connection, application, *, lock_timeout='5s', statement_timeout='60s')

Run the structural tree as the owner, one transaction per revision.

Parameters:

• connection (Connection)

• application (Application)

• lock_timeout (str)

• statement_timeout (str)

Return type:

None

loom.core.repository.sqlalchemy.migrations.scope_protection_hook(application)

Return the process_revision_directives callable for application.

Parameters:

application (Application)

Return type:

RevisionHook