Skip to content

scene_pipelines.core.versioning

Generic DDL schema-version helpers (per MySQL schema / version table).

Each schema defines its SchemaVersion table from SchemaVersionTable, which records the code's version in the same step that creates the table. The helpers below check an existing database against the installed code, e.g.::

from scene_pipelines.core.versioning import ensure_schema_version
from scene_pipelines.schemas.scene._schema import (
    SCENE_SCHEMA_VERSION,
    SchemaVersion,
)

ensure_schema_version(SCENE_SCHEMA_VERSION, SchemaVersion)

SchemaVersionError

Bases: RuntimeError

DB schema definition version does not match installed code.

SchemaVersionStatus dataclass

Comparison of installed code vs the latest version row in the DB.

SchemaVersionTable

Bases: Manual

Base for a schema's SchemaVersion table (subclass it; do not decorate this).

Creating the table records code_version right away, so a database never exists without its version. Append a row when a migration has been applied; the latest applied_at row is the current version.

Attributes:

Name Type Description
code_version str

Schema version of the installed code, e.g. "0.2.0".

declare

declare(context=None)

Create the table and record code_version in it.

assert_database_compatible

assert_database_compatible(database: str) -> None

Check database's recorded schema version against the code, once per process.

Uses the SchemaVersionTable bound to database; a database without one (e.g. a lab's own schema) is not checked.

Raises:

Type Description
SchemaVersionError

If the recorded version is missing or differs.

get_db_schema_version

get_db_schema_version(version_table: type[Manual]) -> str | None

Return the most recently applied schema version, or None if unset.

check_schema_version

check_schema_version(expected_version: str, version_table: type[Manual]) -> SchemaVersionStatus

Compare expected_version (code) to the latest DB version row.

assert_schema_compatible

assert_schema_compatible(expected_version: str, version_table: type[Manual]) -> str

Require DB version to match code; raise SchemaVersionError otherwise.

ensure_schema_version

ensure_schema_version(expected_version: str, version_table: type[Manual], *, notes: str = '') -> str

Record expected_version on an empty DB, or confirm an exact match.