Migrations¶

migrations ¶

Alembic integration behind a project's env.py: URL resolution, import_models, run, revision generation (process_revision_directives). Extra spltz-viur-models[migrations].

target_metadata ¶

target_metadata() -> Any

SQLModel.metadata — the whole schema once the models are imported.

Source code in src/viur/models/migrations.py

def target_metadata() -> t.Any:
    """``SQLModel.metadata`` — the whole schema once the models are imported."""
    return SQLModel.metadata

import_models ¶

import_models(*packages: str) -> list[str]

Import each package and its non-private submodules (one level) to fill SQLModel.metadata. Returns the imported module names.

Source code in src/viur/models/migrations.py

def import_models(*packages: str) -> list[str]:
    """Import each package and its non-private submodules (one level) to fill ``SQLModel.metadata``.
    Returns the imported module names."""
    imported = []
    for package_name in packages:
        package = importlib.import_module(package_name)
        imported.append(package_name)
        for info in pkgutil.iter_modules(getattr(package, "__path__", [])):
            if info.name.startswith("_"):
                continue
            importlib.import_module(f"{package_name}.{info.name}")
            imported.append(f"{package_name}.{info.name}")
    return imported

resolve_url ¶

resolve_url(config: Config | None = None, *, x_args: dict | None = None, fallback_url: str | None = None) -> str

Database URL for this run, most explicit first: -x url=… (x_args), $VIUR_MODELS_DSN, the configured engine, the conf.models preset, fallback_url, sqlalchemy.url from alembic.ini. -x db=<name> selects a non-default database (engine or conf.models.databases entry only). RuntimeError when none applies.

Source code in src/viur/models/migrations.py

def resolve_url(
    config: "Config | None" = None,
    *,
    x_args: dict | None = None,
    fallback_url: str | None = None,
) -> str:
    """Database URL for this run, most explicit first: ``-x url=…`` (``x_args``),
    ``$VIUR_MODELS_DSN``, the configured engine, the ``conf.models`` preset, ``fallback_url``,
    ``sqlalchemy.url`` from ``alembic.ini``. ``-x db=<name>`` selects a non-default database
    (engine or ``conf.models.databases`` entry only). ``RuntimeError`` when none applies."""
    import os

    from . import db

    database = _database(x_args)
    if x_args and (url := (x_args.get("url") or "").strip()):
        return url
    if database == db.DEFAULT and (url := os.environ.get(DSN_ENV_VAR, "").strip()):
        return url

    try:
        engine = db.get_engine(database)
    except RuntimeError:
        pass
    else:
        return str(engine.url.render_as_string(hide_password=False))

    # a set but misconfigured preset must surface: no try/except here
    cfg = _models_conf()
    if database != db.DEFAULT:
        if cfg is None:
            raise RuntimeError(f"No engine {database!r} and conf.models is not installed")
        return db.url_from_conf(database)
    if cfg is not None and db.DEFAULT in cfg.databases:
        return db.url_from_conf()

    if fallback_url and fallback_url.strip():
        return fallback_url.strip()

    if config is not None:
        if url := (config.get_main_option("sqlalchemy.url") or "").strip():
            return url

    raise RuntimeError(
        "No database URL for the migration. Tried, in order: -x url=…, "
        f"${DSN_ENV_VAR}, the configured engine (viur.models.db.configure), "
        "conf.models, the env.py fallback, and sqlalchemy.url in alembic.ini."
    )

include_object ¶

include_object(obj: Any, name: str | None, type_: str, reflected: bool, compare_to: Any) -> bool

Autogenerate filter: everything but alembic_version. Wrap it to exclude foreign tables.

Source code in src/viur/models/migrations.py

def include_object(
    obj: t.Any, name: str | None, type_: str, reflected: bool, compare_to: t.Any,
) -> bool:
    """Autogenerate filter: everything but ``alembic_version``. Wrap it to exclude foreign tables."""
    return not (type_ == "table" and name == "alembic_version")

include_object_for ¶

include_object_for(database: str) -> Callable

include_object restricted to the tables of database (db.tables_for).

Source code in src/viur/models/migrations.py

def include_object_for(database: str) -> t.Callable:
    """``include_object`` restricted to the tables of ``database`` (``db.tables_for``)."""
    from . import db

    names = {table.name for table in db.tables_for(database)}

    def _include(obj: t.Any, name: str | None, type_: str, reflected: bool, compare_to: t.Any) -> bool:
        if type_ == "table" and name not in names:
            return False
        return include_object(obj, name, type_, reflected, compare_to)

    return _include

render_item ¶

render_item(type_: str, obj: Any, autogen_context: Any) -> Any

Render a TypeDecorator as its impl type (RecordJSON cannot round-trip); SQLModel's decorators and dialect-specific impls are left to Alembic.

Source code in src/viur/models/migrations.py

def render_item(type_: str, obj: t.Any, autogen_context: t.Any) -> t.Any:
    """Render a ``TypeDecorator`` as its ``impl`` type (``RecordJSON`` cannot round-trip);
    SQLModel's decorators and dialect-specific impls are left to Alembic."""
    from sqlalchemy.types import TypeDecorator, TypeEngine

    if type_ != "type" or not isinstance(obj, TypeDecorator):
        return False
    if type(obj).__module__.startswith("sqlmodel."):
        return False
    impl = obj.impl if isinstance(obj.impl, TypeEngine) else obj.impl()
    if not type(impl).__module__.startswith("sqlalchemy.sql.sqltypes"):
        return False  # dialect-specific impl (JSONB, …): no sa. name
    autogen_context.imports.add("import sqlalchemy as sa")
    return f"sa.{impl!r}"

run_offline ¶

run_offline(url: str, database: str = 'default', **overrides: Any) -> None

--sql mode.

Source code in src/viur/models/migrations.py

def run_offline(url: str, database: str = "default", **overrides: t.Any) -> None:
    """``--sql`` mode."""
    context.configure(url=url, literal_binds=True,
                      dialect_opts={"paramstyle": "named"},
                      **_configure_kwargs(url, database, **overrides))
    with context.begin_transaction():
        context.run_migrations()

run_online ¶

run_online(url: str, config: Config, database: str = 'default', **overrides: Any) -> None

Connect and run the migrations in one transaction.

Source code in src/viur/models/migrations.py

def run_online(url: str, config: "Config", database: str = "default", **overrides: t.Any) -> None:
    """Connect and run the migrations in one transaction."""
    section = config.get_section(config.config_ini_section) or {}
    section["sqlalchemy.url"] = url
    engine: "Engine" = engine_from_config(
        section, prefix="sqlalchemy.", poolclass=pool.NullPool,
    )
    connection: "Connection"
    with engine.connect() as connection:
        context.configure(connection=connection, **_configure_kwargs(url, database, **overrides))
        with context.begin_transaction():
            context.run_migrations()
    engine.dispose()

run ¶

run(*, fallback_url: str | None = None, **overrides: Any) -> str

env.py entry point: resolve the URL (-x db=<name> picks the database), dispatch offline/online; overrides go to context.configure. Returns the URL.

Source code in src/viur/models/migrations.py

def run(*, fallback_url: str | None = None, **overrides: t.Any) -> str:
    """``env.py`` entry point: resolve the URL (``-x db=<name>`` picks the database), dispatch
    offline/online; ``overrides`` go to ``context.configure``. Returns the URL."""
    config = context.config
    x_args = context.get_x_argument(as_dictionary=True)
    url = resolve_url(config, x_args=x_args, fallback_url=fallback_url)
    database = _database(x_args)
    if context.is_offline_mode():
        run_offline(url, database, **overrides)
    else:
        run_online(url, config, database, **overrides)
    return url

process_revision_directives ¶

process_revision_directives(context: Any, revision: Any, directives: list) -> None

Alembic hook: correct the DDL diff for bone-level transitions, write the structure snapshot.

Source code in src/viur/models/migrations.py

def process_revision_directives(context: t.Any, revision: t.Any, directives: list) -> None:
    """Alembic hook: correct the DDL diff for bone-level transitions, write the structure snapshot."""
    from alembic import util as alembic_util

    from . import schema

    if not directives:
        return
    script = directives[0]
    if script.upgrade_ops is None:
        return

    config = context.config if hasattr(context, "config") else None
    location = config.get_main_option("script_location") if config else None
    if not location:
        return
    directory = schema.snapshot_dir(location)

    current = schema.snapshot()
    previous = schema.load(directory, _parent_revision(revision))
    transitions = schema.diff(previous, current)

    for name in _name_constraints(script.upgrade_ops):
        alembic_util.msg(f"viur-models: named the constraint {name} (SQLite needs it)")

    if transitions:
        for note in _generate_transition_ops(script.upgrade_ops, transitions):
            alembic_util.msg(f"viur-models: {note}")

    if _is_dry_run(context):
        return  # alembic check

    schema.save(directory, script.rev_id, current)
    schema.prune(directory, _known_revisions(config) | {script.rev_id})

scaffold ¶

Alembic scaffold generator (alembic.ini + migrations/); never overwrites.

generate ¶

generate(root: str | PathLike, *, app_dir: str | PathLike | None = None, models_package: str = 'models', settings_module: str = 'models_db') -> list[str]

Write the missing scaffold files into root.

Parameters:

Name	Type	Description	Default
`app_dir`	`str \| PathLike \| None`	Distribution folder for `prepend_sys_path` (default: cwd).	`None`
`models_package`	`str`	Package `env.py` imports to fill `SQLModel.metadata`.	`'models'`
`settings_module`	`str`	Module providing the `url()` fallback.	`'models_db'`

Returns:

Type	Description
`list[str]`	Created paths relative to `root`.

Source code in src/viur/models/scaffold.py

def generate(
    root: str | os.PathLike,
    *,
    app_dir: str | os.PathLike | None = None,
    models_package: str = "models",
    settings_module: str = "models_db",
) -> list[str]:
    """Write the missing scaffold files into ``root``.

    :param app_dir: Distribution folder for ``prepend_sys_path`` (default: cwd).
    :param models_package: Package ``env.py`` imports to fill ``SQLModel.metadata``.
    :param settings_module: Module providing the ``url()`` fallback.
    :returns: Created paths relative to ``root``.
    """
    root = pathlib.Path(root).resolve()
    app_dir = pathlib.Path(app_dir).resolve() if app_dir else pathlib.Path.cwd()
    fields = {
        "migrations_dirname": MIGRATIONS_DIRNAME,
        "models_package": models_package,
        "settings_module": settings_module,
        "sys_path": _sys_path_entry(root, app_dir),
    }
    contents = {
        "alembic.ini": ALEMBIC_INI.format(**fields),
        f"{MIGRATIONS_DIRNAME}/env.py": ENV_PY.format(**fields),
        f"{MIGRATIONS_DIRNAME}/script.py.mako": _script_mako(),
        f"{MIGRATIONS_DIRNAME}/README": README,
        f"{MIGRATIONS_DIRNAME}/{VERSIONS_DIRNAME}/.gitkeep": "",
    }

    created = []
    for name in FILES:
        path = root / name
        if path.exists():
            continue
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(contents[name], encoding="utf-8")
        created.append(name)
    return created

is_complete ¶

is_complete(root: str | PathLike) -> bool

Whether root holds every scaffold file.

Source code in src/viur/models/scaffold.py

def is_complete(root: str | os.PathLike) -> bool:
    """Whether ``root`` holds every scaffold file."""
    root = pathlib.Path(root)
    return all((root / name).exists() for name in FILES)

schema ¶

Structure snapshots and bone-level diffing; snapshots live in <script_location>/structures/<revision>.json.

Transition `dataclass` ¶

One bone-level change.

Parameters:

Name	Type	Description	Default
`kind`	`str`	See `diff`.	required
`old`	`dict \| None`	Previous bone entry (`None` for additions).	`None`
`new`	`dict \| None`	Current bone entry (`None` for removals).	`None`
`detail`	`dict`	Facts the operation needs (link table, columns, languages, …).	`dict()`

Source code in src/viur/models/schema.py

@dataclasses.dataclass(frozen=True)
class Transition:
    """One bone-level change.

    :param kind: See ``diff``.
    :param old: Previous bone entry (``None`` for additions).
    :param new: Current bone entry (``None`` for removals).
    :param detail: Facts the operation needs (link table, columns, languages, …).
    """

    kind: str
    table: str
    field: str
    old: dict | None = None
    new: dict | None = None
    detail: dict = dataclasses.field(default_factory=dict)

describe ¶

describe(model: type) -> dict

Snapshot entry of one model; resolve_refs=False (no skeleton registry in a shell), bypasses the structure cache.

Source code in src/viur/models/schema.py

def describe(model: type) -> dict:
    """Snapshot entry of one model; ``resolve_refs=False`` (no skeleton registry in a shell),
    bypasses the structure cache."""
    from .structure import structure_for_model

    relations = {}
    for rel_name, info in model.viur_relations().items():
        relations[rel_name] = _relation_shape(model, rel_name, info)
    return {
        "table": model._viur_kind(),
        "structure": structure_for_model(model, resolve_refs=False),
        "relations": relations,
    }

snapshot ¶

snapshot() -> dict

Describe every imported Model table (migrations.import_models).

Source code in src/viur/models/schema.py

def snapshot() -> dict:
    """Describe every imported ``Model`` table (``migrations.import_models``)."""
    return {
        model._viur_kind(): describe(model)
        for model in sorted(_table_models(), key=lambda cls: cls._viur_kind())
    }

snapshot_dir ¶

snapshot_dir(script_location: str | Path) -> Path

Snapshot directory of an Alembic script directory.

Source code in src/viur/models/schema.py

def snapshot_dir(script_location: str | pathlib.Path) -> pathlib.Path:
    """Snapshot directory of an Alembic script directory."""
    return pathlib.Path(script_location) / "structures"

prune ¶

prune(directory: str | Path, keep: Iterable[str]) -> list[str]

Delete snapshots of revisions that no longer exist. Returns the removed ids.

Source code in src/viur/models/schema.py

def prune(directory: str | pathlib.Path, keep: t.Iterable[str]) -> list[str]:
    """Delete snapshots of revisions that no longer exist. Returns the removed ids."""
    path = pathlib.Path(directory)
    if not path.is_dir():
        return []
    keep = set(keep)
    removed = []
    for snapshot_file in sorted(path.glob("*.json")):
        if snapshot_file.stem not in keep:
            snapshot_file.unlink()
            removed.append(snapshot_file.stem)
    return removed

load ¶

load(directory: str | Path, revision: str | None) -> dict

One snapshot; {} for an unknown/missing revision (diff then reports only additions).

Source code in src/viur/models/schema.py

def load(directory: str | pathlib.Path, revision: str | None) -> dict:
    """One snapshot; ``{}`` for an unknown/missing revision (``diff`` then reports only additions)."""
    if not revision:
        return {}
    path = pathlib.Path(directory) / f"{revision}.json"
    if not path.is_file():
        return {}
    try:
        return json.loads(path.read_text())
    except ValueError:
        return {}

diff ¶

diff(old: dict, new: dict) -> list[Transition]

Bone-level transitions between two snapshots, by table then field.

Kinds: field_added/field_removed, multiple_collapsed/multiple_expanded (link table ↔ FK column), languages_reduced/languages_expanded, type_changed (detail["from"]/["to"]), select_values_changed, precision_changed, using_field_added/using_field_removed (link model payload).

Source code in src/viur/models/schema.py

def diff(old: dict, new: dict) -> list[Transition]:
    """Bone-level transitions between two snapshots, by table then field.

    Kinds: ``field_added``/``field_removed``, ``multiple_collapsed``/``multiple_expanded``
    (link table ↔ FK column), ``languages_reduced``/``languages_expanded``, ``type_changed``
    (``detail["from"]``/``["to"]``), ``select_values_changed``, ``precision_changed``,
    ``using_field_added``/``using_field_removed`` (link model payload).
    """
    transitions: list[Transition] = []

    for table in sorted(new):
        current, previous = new[table], old.get(table)
        if previous is None:
            continue  # new table: plain CREATE TABLE
        old_structure = previous.get("structure", {})
        new_structure = current.get("structure", {})
        old_relations = previous.get("relations", {})
        new_relations = current.get("relations", {})

        for field in sorted(set(old_structure) | set(new_structure)):
            old_bone = old_structure.get(field)
            new_bone = new_structure.get(field)

            if old_bone is None:
                transitions.append(Transition(
                    "field_added", table, field, None, new_bone,
                    {"relation": field in new_relations},
                ))
                continue
            if new_bone is None:
                transitions.append(Transition("field_removed", table, field, old_bone, None))
                continue
            if not _bone_changed(old_bone, new_bone):
                # unchanged bone; its using payload may differ
                transitions.extend(_using_transitions(
                    table, field, old_relations.get(field), new_relations.get(field),
                    new_bone=new_bone,
                ))
                continue

            old_multiple = bool(old_bone.get("multiple"))
            new_multiple = bool(new_bone.get("multiple"))
            is_relation = field in old_relations or field in new_relations

            if is_relation and old_multiple != new_multiple:
                old_shape = old_relations.get(field, {})
                new_shape = new_relations.get(field, {})
                if old_multiple:
                    transitions.append(Transition(
                        "multiple_collapsed", table, field, old_bone, new_bone,
                        {
                            "link_table": old_shape.get("link_table"),
                            "link_parent_fk": old_shape.get("link_parent_fk"),
                            "link_dest_fk": old_shape.get("link_dest_fk"),
                            "target_column": new_shape.get("fk"),
                            "target_table": new_shape.get("target_table")
                            or old_shape.get("target_table"),
                        },
                    ))
                else:
                    transitions.append(Transition(
                        "multiple_expanded", table, field, old_bone, new_bone,
                        {
                            "link_table": new_shape.get("link_table"),
                            "link_parent_fk": new_shape.get("link_parent_fk"),
                            "link_dest_fk": new_shape.get("link_dest_fk"),
                            "source_column": old_shape.get("fk"),
                        },
                    ))
                continue

            old_langs, new_langs = old_bone.get("languages"), new_bone.get("languages")
            if bool(old_langs) != bool(new_langs):
                if old_langs:
                    transitions.append(Transition(
                        "languages_reduced", table, field, old_bone, new_bone,
                        {"languages": list(old_langs)},
                    ))
                else:
                    transitions.append(Transition(
                        "languages_expanded", table, field, old_bone, new_bone,
                        {"languages": list(new_langs)},
                    ))
                continue

            if old_bone.get("type") != new_bone.get("type"):
                transitions.append(Transition(
                    "type_changed", table, field, old_bone, new_bone,
                    {"from": old_bone.get("type"), "to": new_bone.get("type")},
                ))
                continue

            if new_bone.get("type", "").startswith("select"):
                old_values = set((old_bone.get("values") or {}))
                new_values = set((new_bone.get("values") or {}))
                if old_values != new_values:
                    transitions.append(Transition(
                        "select_values_changed", table, field, old_bone, new_bone,
                        {
                            "added": sorted(new_values - old_values),
                            "removed": sorted(old_values - new_values),
                        },
                    ))
                    continue

            if new_bone.get("type", "").startswith("numeric") \
                    and _numeric_precision(old_bone) != _numeric_precision(new_bone):
                transitions.append(Transition(
                    "precision_changed", table, field, old_bone, new_bone,
                    {"precision": new_bone.get("precision", 0)},
                ))
                continue

            transitions.extend(_using_transitions(
                table, field, old_relations.get(field), new_relations.get(field),
                new_bone=new_bone,
            ))

        transitions = _pair_renamed_relations(
            transitions, table, old_relations, new_relations,
        )

    return transitions

migrate ¶

Alembic operations for bone-level data migrations; row-wise coercions per viur-core's rules.