55 lines
2.3 KiB
Python
55 lines
2.3 KiB
Python
"""add durable Campaign child Run identity
|
|
|
|
Revision ID: c2f4a6b8d0e1
|
|
Revises: 99dbae2a20da
|
|
Create Date: 2026-08-06
|
|
"""
|
|
|
|
from typing import Sequence, Union
|
|
|
|
import sqlalchemy as sa
|
|
import sqlmodel # noqa: F401
|
|
from alembic import op
|
|
|
|
|
|
revision: str = "c2f4a6b8d0e1"
|
|
down_revision: Union[str, Sequence[str], None] = "99dbae2a20da"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
"""Add nullable identity columns without backfilling historical Runs."""
|
|
# Adding nullable columns directly avoids rebuilding ``eval_runs``. That
|
|
# table participates in a foreign-key cycle with ``campaigns`` and
|
|
# ``eval_targets``; SQLite batch recreation can otherwise fail while
|
|
# topologically sorting the copied table. A unique index has the same
|
|
# database-level idempotency semantics as a composite unique constraint.
|
|
inspector = sa.inspect(op.get_bind())
|
|
columns = {column["name"] for column in inspector.get_columns("eval_runs")}
|
|
if "campaign_plan_index" not in columns:
|
|
op.add_column("eval_runs", sa.Column("campaign_plan_index", sa.Integer(), nullable=True))
|
|
if "campaign_occurrence_index" not in columns:
|
|
op.add_column("eval_runs", sa.Column("campaign_occurrence_index", sa.Integer(), nullable=True))
|
|
index_names = {index["name"] for index in inspector.get_indexes("eval_runs")}
|
|
if "uq_eval_runs_campaign_occurrence" not in index_names:
|
|
op.create_index(
|
|
"uq_eval_runs_campaign_occurrence",
|
|
"eval_runs",
|
|
["campaign_id", "campaign_plan_index", "campaign_occurrence_index"],
|
|
unique=True,
|
|
)
|
|
|
|
|
|
def downgrade() -> None:
|
|
"""Remove the child identity fields and their uniqueness constraint."""
|
|
inspector = sa.inspect(op.get_bind())
|
|
index_names = {index["name"] for index in inspector.get_indexes("eval_runs")}
|
|
if "uq_eval_runs_campaign_occurrence" in index_names:
|
|
op.drop_index("uq_eval_runs_campaign_occurrence", table_name="eval_runs")
|
|
columns = {column["name"] for column in inspector.get_columns("eval_runs")}
|
|
if "campaign_occurrence_index" in columns:
|
|
op.drop_column("eval_runs", "campaign_occurrence_index")
|
|
if "campaign_plan_index" in columns:
|
|
op.drop_column("eval_runs", "campaign_plan_index")
|