script-build-and-framework-tmp/script_build_host/migrations/versions/0004_candidate_source_identity.py

"""Persist source-qualified logical identities for candidate workspace seeds.

Revision ID: 0004_candidate_source_identity
Revises: 0003_phase_three_publication
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import mysql

revision: str = "0004_candidate_source_identity"
down_revision: str | None = "0003_phase_three_publication"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
    timestamp = sa.DateTime(timezone=True).with_variant(mysql.DATETIME(fsp=6), "mysql")
    op.create_table(
        "script_build_candidate_source_identity",
        sa.Column("id", sa.BigInteger().with_variant(sa.Integer, "sqlite"), primary_key=True),
        sa.Column("workspace_artifact_version_id", sa.BigInteger(), nullable=False),
        sa.Column("entity_kind", sa.String(32), nullable=False),
        sa.Column("target_local_id", sa.BigInteger(), nullable=False),
        sa.Column("source_artifact_version_id", sa.BigInteger(), nullable=False),
        sa.Column("source_artifact_digest", sa.String(64), nullable=False),
        sa.Column("source_local_id", sa.BigInteger(), nullable=False),
        sa.Column("created_at", timestamp, nullable=False),
        sa.UniqueConstraint(
            "workspace_artifact_version_id",
            "entity_kind",
            "target_local_id",
            name="uq_candidate_source_target",
        ),
        sa.UniqueConstraint(
            "workspace_artifact_version_id",
            "source_artifact_version_id",
            "entity_kind",
            "source_local_id",
            name="uq_candidate_source_origin",
        ),
        sa.CheckConstraint(
            "entity_kind IN ('paragraph', 'element')", name="ck_candidate_source_kind"
        ),
        sa.CheckConstraint(
            "length(source_artifact_digest) = 64", name="ck_candidate_source_digest"
        ),
    )
    op.create_index(
        "ix_candidate_source_workspace",
        "script_build_candidate_source_identity",
        ["workspace_artifact_version_id"],
    )


def downgrade() -> None:
    op.drop_index(
        "ix_candidate_source_workspace", table_name="script_build_candidate_source_identity"
    )
    op.drop_table("script_build_candidate_source_identity")