SupplyAgent/alembic/versions/20260731_10_platform_demand_mvp.py

"""add evidence packages and versioned platform demand objects

Revision ID: 20260731_10
Revises: 20260731_09
Create Date: 2026-07-31
"""
from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "20260731_10"
down_revision: str | None = "20260731_09"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def _pipeline_table_options() -> dict[str, str]:
    bind = op.get_bind()
    if bind.dialect.name != "mysql":
        return {}
    row = bind.execute(
        sa.text(
            "SELECT CHARACTER_SET_NAME, COLLATION_NAME "
            "FROM INFORMATION_SCHEMA.COLUMNS "
            "WHERE TABLE_SCHEMA = DATABASE() "
            "AND TABLE_NAME = 'pipeline_run' "
            "AND COLUMN_NAME = 'run_id'"
        )
    ).one()
    return {
        "mysql_charset": str(row.CHARACTER_SET_NAME),
        "mysql_collate": str(row.COLLATION_NAME),
    }


def upgrade() -> None:
    table_options = _pipeline_table_options()
    op.create_table(
        "evidence_package",
        sa.Column("package_id", sa.String(36), primary_key=True),
        sa.Column(
            "run_id",
            sa.String(36),
            sa.ForeignKey("pipeline_run.run_id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column("biz_dt", sa.String(8), nullable=False),
        sa.Column("package_version", sa.Integer(), nullable=False),
        sa.Column("status", sa.String(24), nullable=False),
        sa.Column("source_snapshot_hash", sa.String(64), nullable=False),
        sa.Column("source_versions_json", sa.JSON(), nullable=False),
        sa.Column("evidence_count", sa.Integer(), nullable=False),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.UniqueConstraint("run_id", name="uk_evidence_package_run"),
        **table_options,
    )
    op.create_index(
        "idx_evidence_package_biz",
        "evidence_package",
        ["biz_dt", "created_at"],
    )

    op.create_table(
        "raw_demand_expression",
        sa.Column("expression_id", sa.String(36), primary_key=True),
        sa.Column(
            "package_id",
            sa.String(36),
            sa.ForeignKey("evidence_package.package_id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column("source_type", sa.String(64), nullable=False),
        sa.Column("source_record_id", sa.String(128), nullable=False),
        sa.Column("raw_text", sa.String(512), nullable=False),
        sa.Column("original_payload_json", sa.JSON(), nullable=False),
        sa.Column("content_hash", sa.String(64), nullable=False),
        sa.Column("observed_biz_dt", sa.String(8), nullable=False),
        sa.Column("data_quality", sa.String(24), nullable=False),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.UniqueConstraint(
            "package_id",
            "source_type",
            "source_record_id",
            name="uk_raw_expression_source",
        ),
        **table_options,
    )
    op.create_index(
        "idx_raw_expression_package",
        "raw_demand_expression",
        ["package_id"],
    )
    op.create_index(
        "idx_raw_expression_hash",
        "raw_demand_expression",
        ["content_hash"],
    )

    op.create_table(
        "standard_demand_term",
        sa.Column("term_id", sa.String(36), primary_key=True),
        sa.Column("canonical_key", sa.String(64), nullable=False),
        sa.Column("canonical_text", sa.String(256), nullable=False),
        sa.Column("normalized_text", sa.String(256), nullable=False),
        sa.Column("status", sa.String(24), nullable=False),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.UniqueConstraint("canonical_key", name="uk_standard_demand_term_key"),
        **table_options,
    )
    op.create_index(
        "idx_standard_demand_term_text",
        "standard_demand_term",
        ["normalized_text"],
    )

    op.create_table(
        "standard_demand_alias",
        sa.Column("alias_id", sa.String(36), primary_key=True),
        sa.Column(
            "term_id",
            sa.String(36),
            sa.ForeignKey("standard_demand_term.term_id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column(
            "expression_id",
            sa.String(36),
            sa.ForeignKey("raw_demand_expression.expression_id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column("alias_text", sa.String(512), nullable=False),
        sa.Column("relation_type", sa.String(32), nullable=False),
        sa.Column("confidence", sa.Numeric(6, 5), nullable=False),
        sa.Column("reason", sa.Text(), nullable=False),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.UniqueConstraint(
            "term_id",
            "expression_id",
            name="uk_standard_demand_alias_expression",
        ),
        **table_options,
    )
    op.create_index(
        "idx_standard_demand_alias_term",
        "standard_demand_alias",
        ["term_id"],
    )

    op.create_table(
        "platform_demand",
        sa.Column("platform_demand_id", sa.String(36), primary_key=True),
        sa.Column("canonical_key", sa.String(64), nullable=False),
        sa.Column("name", sa.String(256), nullable=False),
        sa.Column("description", sa.Text(), nullable=False),
        sa.Column("status", sa.String(24), nullable=False),
        sa.Column("lifecycle_state", sa.String(24), nullable=False),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.UniqueConstraint("canonical_key", name="uk_platform_demand_key"),
        **table_options,
    )
    op.create_index("idx_platform_demand_name", "platform_demand", ["name"])
    op.create_index("idx_platform_demand_status", "platform_demand", ["status"])

    op.create_table(
        "platform_demand_version",
        sa.Column("platform_demand_version_id", sa.String(36), primary_key=True),
        sa.Column(
            "platform_demand_id",
            sa.String(36),
            sa.ForeignKey("platform_demand.platform_demand_id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column(
            "term_id",
            sa.String(36),
            sa.ForeignKey("standard_demand_term.term_id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column(
            "package_id",
            sa.String(36),
            sa.ForeignKey("evidence_package.package_id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column(
            "run_id",
            sa.String(36),
            sa.ForeignKey("pipeline_run.run_id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column("source_demand_grade_id", sa.BigInteger(), nullable=False),
        sa.Column("version_no", sa.Integer(), nullable=False),
        sa.Column("biz_dt", sa.String(8), nullable=False),
        sa.Column("name", sa.String(256), nullable=False),
        sa.Column("description", sa.Text(), nullable=False),
        sa.Column("cognition_confidence", sa.Numeric(6, 5), nullable=False),
        sa.Column("reason", sa.Text(), nullable=False),
        sa.Column("change_type", sa.String(24), nullable=False),
        sa.Column("evidence_hash", sa.String(64), nullable=False),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.UniqueConstraint(
            "platform_demand_id",
            "version_no",
            name="uk_platform_demand_version_no",
        ),
        sa.UniqueConstraint(
            "platform_demand_id",
            "run_id",
            name="uk_platform_demand_version_run",
        ),
        **table_options,
    )
    op.create_index(
        "idx_platform_demand_version_biz",
        "platform_demand_version",
        ["biz_dt", "run_id"],
    )

    op.create_table(
        "platform_demand_category_rel",
        sa.Column("rel_id", sa.String(36), primary_key=True),
        sa.Column(
            "platform_demand_version_id",
            sa.String(36),
            sa.ForeignKey(
                "platform_demand_version.platform_demand_version_id",
                ondelete="RESTRICT",
            ),
            nullable=False,
        ),
        sa.Column("category_id", sa.BigInteger(), nullable=False),
        sa.Column("relation_type", sa.String(32), nullable=False),
        sa.Column("relation_source", sa.String(64), nullable=False),
        sa.Column("reason", sa.Text(), nullable=False),
        sa.Column("confidence", sa.Numeric(6, 5), nullable=False),
        sa.Column("is_inferred", sa.Boolean(), nullable=False),
        sa.Column("status", sa.String(24), nullable=False),
        sa.Column("valid_from_biz_dt", sa.String(8), nullable=False),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.UniqueConstraint(
            "platform_demand_version_id",
            "category_id",
            "relation_type",
            name="uk_platform_demand_category_rel",
        ),
        **table_options,
    )
    op.create_index(
        "idx_platform_demand_category_node",
        "platform_demand_category_rel",
        ["category_id"],
    )


def downgrade() -> None:
    op.drop_index(
        "idx_platform_demand_category_node",
        table_name="platform_demand_category_rel",
    )
    op.drop_table("platform_demand_category_rel")
    op.drop_index(
        "idx_platform_demand_version_biz",
        table_name="platform_demand_version",
    )
    op.drop_table("platform_demand_version")
    op.drop_index("idx_platform_demand_status", table_name="platform_demand")
    op.drop_index("idx_platform_demand_name", table_name="platform_demand")
    op.drop_table("platform_demand")
    op.drop_index(
        "idx_standard_demand_alias_term",
        table_name="standard_demand_alias",
    )
    op.drop_table("standard_demand_alias")
    op.drop_index(
        "idx_standard_demand_term_text",
        table_name="standard_demand_term",
    )
    op.drop_table("standard_demand_term")
    op.drop_index("idx_raw_expression_hash", table_name="raw_demand_expression")
    op.drop_index("idx_raw_expression_package", table_name="raw_demand_expression")
    op.drop_table("raw_demand_expression")
    op.drop_index("idx_evidence_package_biz", table_name="evidence_package")
    op.drop_table("evidence_package")