"""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")