"""Add V2 product import mapper session and mapping dictionary tables. Revision ID: 0069_add_v2_import_mapper_tables Revises: 0068_unique_active_collection_path_slug Create Date: 2026-08-21 18:20:00.000000 """ from __future__ import annotations import sqlalchemy as sa from alembic import op from sqlalchemy.dialects import postgresql revision = "0069_add_v2_import_mapper_tables" down_revision = "0068_unique_active_collection_path_slug" branch_labels = None depends_on = None def upgrade() -> None: op.drop_constraint("v2_staging_batch_status_valid", "v2_staging_batch", type_="check") op.create_check_constraint( "v2_staging_batch_status_valid", "v2_staging_batch", "status IN (" "'processing','needs_review','promotable','completed'," "'uploaded','analyzing','needs_mapping','ready','importing','failed','cancelled'" ")", ) op.add_column( "v2_staging_batch", sa.Column("source_profile", sa.String(length=128), nullable=False, server_default=""), ) op.add_column("v2_staging_batch", sa.Column("error_message", sa.Text(), nullable=True)) op.add_column( "v2_staging_batch", sa.Column("summary_json", postgresql.JSONB(astext_type=sa.Text()), nullable=True), ) op.add_column( "v2_staging_batch", sa.Column("result_json", postgresql.JSONB(astext_type=sa.Text()), nullable=True), ) op.add_column( "v2_staging_batch", sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")), ) op.create_table( "v2_import_mapping", sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True), sa.Column( "batch_id", postgresql.UUID(as_uuid=False), sa.ForeignKey("v2_staging_batch.id", ondelete="CASCADE"), nullable=False, ), sa.Column("entity_type", sa.String(length=32), nullable=False), sa.Column("source_value", sa.Text(), nullable=False), sa.Column("normalized_value", sa.Text(), nullable=False), sa.Column("parent_key", sa.Text(), nullable=False, server_default=""), sa.Column("parent_entity_type", sa.String(length=32), nullable=True), sa.Column("parent_source_value", sa.Text(), nullable=True), sa.Column("parent_target_id", sa.BigInteger(), nullable=True), sa.Column("occurrence_count", sa.Integer(), nullable=False, server_default=sa.text("0")), sa.Column("status", sa.String(length=32), nullable=False, server_default="unmatched"), sa.Column("matched_target_id", sa.BigInteger(), nullable=True), sa.Column("matched_target_name", sa.Text(), nullable=True), sa.Column("suggested_targets", postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column("create_payload", postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column("resolution_source", sa.String(length=64), nullable=True), sa.Column("resolved_by", sa.Text(), nullable=True), sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True), sa.Column("is_optional", sa.Boolean(), nullable=False, server_default=sa.text("false")), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")), sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")), sa.CheckConstraint( "entity_type IN ('brand','hub','category','collection','attribute','attribute_option')", name="v2_import_mapping_entity_type_valid", ), sa.CheckConstraint( "status IN (" "'exact_match','known_mapping','suggested','ambiguous','unmatched','invalid'," "'mapped','created','ignored'" ")", name="v2_import_mapping_status_valid", ), sa.UniqueConstraint( "batch_id", "entity_type", "normalized_value", "parent_key", name="uq_v2_import_mapping_batch_entity_value", ), ) op.create_index("ix_v2_import_mapping_batch_status", "v2_import_mapping", ["batch_id", "status"]) op.create_index("ix_v2_import_mapping_batch_entity", "v2_import_mapping", ["batch_id", "entity_type"]) op.create_table( "v2_import_mapping_dictionary", sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True), sa.Column("entity_type", sa.String(length=32), nullable=False), sa.Column("source_profile", sa.String(length=128), nullable=False, server_default=""), sa.Column("source_value", sa.Text(), nullable=False), sa.Column("normalized_value", sa.Text(), nullable=False), sa.Column("parent_key", sa.Text(), nullable=False, server_default=""), sa.Column("target_id", sa.BigInteger(), nullable=False), sa.Column("target_name", sa.Text(), nullable=False), sa.Column("created_by", sa.Text(), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")), sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True), sa.CheckConstraint( "entity_type IN ('brand','hub','category','collection','attribute','attribute_option')", name="v2_import_mapping_dictionary_entity_type_valid", ), sa.UniqueConstraint( "entity_type", "source_profile", "normalized_value", "parent_key", name="uq_v2_import_mapping_dictionary_scope", ), ) op.create_index( "ix_v2_import_mapping_dictionary_lookup", "v2_import_mapping_dictionary", ["entity_type", "normalized_value"], ) def downgrade() -> None: op.drop_index("ix_v2_import_mapping_dictionary_lookup", table_name="v2_import_mapping_dictionary") op.drop_table("v2_import_mapping_dictionary") op.drop_index("ix_v2_import_mapping_batch_entity", table_name="v2_import_mapping") op.drop_index("ix_v2_import_mapping_batch_status", table_name="v2_import_mapping") op.drop_table("v2_import_mapping") op.drop_column("v2_staging_batch", "updated_at") op.drop_column("v2_staging_batch", "result_json") op.drop_column("v2_staging_batch", "summary_json") op.drop_column("v2_staging_batch", "error_message") op.drop_column("v2_staging_batch", "source_profile") op.drop_constraint("v2_staging_batch_status_valid", "v2_staging_batch", type_="check") op.create_check_constraint( "v2_staging_batch_status_valid", "v2_staging_batch", "status IN ('processing','needs_review','promotable','completed')", )