zhifu-platform/migrations/versions/0017_storefront.py

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

revision = "0017_storefront"
down_revision = "0016_platform"
branch_labels = None
depends_on = None


def upgrade():
    op.create_table(
        "catalog_categories",
        sa.Column("tenant_id", mysql.BINARY(16), primary_key=True),
        sa.Column("store_id", mysql.BINARY(16), primary_key=True),
        sa.Column("code", sa.String(32), primary_key=True),
        sa.Column("name", sa.String(64), nullable=False),
        sa.Column("position", mysql.BIGINT(unsigned=True), nullable=False),
        sa.ForeignKeyConstraint(["tenant_id", "store_id"], ["stores.tenant_id", "stores.id"]),
        sa.UniqueConstraint("tenant_id", "store_id", "code", name="uq_catalog_category_code"),
    )
    op.create_table(
        "catalog_products",
        sa.Column("tenant_id", mysql.BINARY(16), primary_key=True),
        sa.Column("store_id", mysql.BINARY(16), primary_key=True),
        sa.Column("id", sa.String(64), primary_key=True),
        sa.Column("category_code", sa.String(32), nullable=False),
        sa.Column("name", sa.String(128), nullable=False),
        sa.Column("description", sa.String(4000), nullable=False),
        sa.Column("specifications", sa.JSON(), nullable=False),
        sa.Column("delivery", sa.String(500), nullable=False),
        sa.Column("image_url", sa.String(1024), nullable=False),
        sa.Column("image_source", sa.String(1024), nullable=False),
        sa.Column("status", sa.String(16), nullable=False),
        sa.ForeignKeyConstraint(
            ["tenant_id", "store_id", "category_code"],
            ["catalog_categories.tenant_id", "catalog_categories.store_id",
             "catalog_categories.code"],
        ),
        sa.CheckConstraint("status IN ('ACTIVE','HIDDEN')", name="status"),
    )
    op.create_table(
        "identity_challenges",
        sa.Column("id", mysql.BINARY(16), primary_key=True),
        sa.Column("tenant_id", mysql.BINARY(16), nullable=False),
        sa.Column("store_id", mysql.BINARY(16), nullable=False),
        sa.Column("customer_id", mysql.BINARY(16), nullable=False),
        sa.Column("request_key", sa.String(64, collation="utf8mb4_bin"), nullable=False),
        sa.Column("channel", sa.String(32), nullable=False),
        sa.Column("status", sa.String(16), nullable=False),
        sa.Column("external_subject", sa.String(128, collation="utf8mb4_bin"), nullable=True),
        sa.Column("display_name", sa.String(128), nullable=True),
        sa.Column("verified_by", mysql.BINARY(16), nullable=True),
        sa.Column("expires_at", mysql.DATETIME(fsp=6), nullable=False),
        sa.Column("created_at", mysql.DATETIME(fsp=6), nullable=False),
        sa.ForeignKeyConstraint(["tenant_id", "store_id"], ["stores.tenant_id", "stores.id"]),
        sa.ForeignKeyConstraint(
            ["tenant_id", "customer_id"], ["principals.tenant_id", "principals.id"]
        ),
        sa.UniqueConstraint(
            "tenant_id", "customer_id", "request_key", name="uq_identity_challenge_key"
        ),
        sa.CheckConstraint("status IN ('PENDING','VERIFIED','LINKED')", name="status"),
    )
    op.create_table(
        "channel_identities",
        sa.Column("tenant_id", mysql.BINARY(16), primary_key=True),
        sa.Column("store_id", mysql.BINARY(16), primary_key=True),
        sa.Column("channel", sa.String(32), primary_key=True),
        sa.Column("external_subject", sa.String(128, collation="utf8mb4_bin"), primary_key=True),
        sa.Column("customer_id", mysql.BINARY(16), nullable=False),
        sa.Column("display_name", sa.String(128), nullable=False),
        sa.Column("challenge_id", mysql.BINARY(16), nullable=False),
        sa.Column("created_at", mysql.DATETIME(fsp=6), nullable=False),
        sa.ForeignKeyConstraint(["tenant_id", "store_id"], ["stores.tenant_id", "stores.id"]),
        sa.ForeignKeyConstraint(
            ["tenant_id", "customer_id"], ["principals.tenant_id", "principals.id"]
        ),
        sa.UniqueConstraint(
            "tenant_id", "store_id", "channel", "customer_id", name="uq_channel_customer"
        ),
    )


def downgrade():
    op.drop_table("channel_identities")
    op.drop_table("identity_challenges")
    op.drop_table("catalog_products")
    op.drop_table("catalog_categories")