"""Data-access request and request-item semantic state. Revision ID: 330 Revises: 329 Create Date: 2026-06-10 DAX-008 adds the request-side durable intent tables. The rows represent approved-intent candidates and Feishu callback staleness snapshots only; they do not create bindings, compile effects, or mutate BigQuery. Principal identity follows the V1 split: * ``principal_email_group_id`` references ``email_groups.id`` and is the semantic Google Group principal. * ``principal_google_group_id_snapshot`` and ``principal_email_snapshot`` are callback-staleness/provider-member snapshots. * aliases are not stored. Request items reference ``data_access_resources.id`` and snapshot dataset locator/source/owner context so callbacks can fail closed if the resource changed while Feishu approval was pending. """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import ( id_type_for, normalize_fk_column_collation, table_kw_for, ) revision = "333" down_revision = "332" branch_labels = None depends_on = None REQUEST_TABLE = "data_access_requests" ITEM_TABLE = "data_access_request_items" def _table_exists(bind, table: str) -> bool: return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1", ), {"t": table}, ).scalar()) def upgrade() -> None: bind = op.get_bind() kw = table_kw_for("identity_persons") if not _table_exists(bind, REQUEST_TABLE): request_id = id_type_for("identity_persons") person_id = id_type_for("identity_persons") group_id = id_type_for("email_groups") entitlement_id = id_type_for("data_access_entitlements") approval_request_id = id_type_for("approval_requests") op.create_table( REQUEST_TABLE, sa.Column("id", request_id, primary_key=True), sa.Column("requester_person_id", person_id, nullable=False), sa.Column("principal_email_group_id", group_id, nullable=False), sa.Column( "principal_google_group_id_snapshot", sa.String(128), nullable=True, ), sa.Column("principal_email_snapshot", sa.String(255), nullable=False), sa.Column("entitlement_id", entitlement_id, nullable=False), sa.Column( "entitlement_snapshot", sa.String(32), nullable=False, server_default="read", ), sa.Column( "entitlement_version_snapshot", sa.String(16), nullable=False, server_default="v1", ), sa.Column( "approval_flow_key", sa.String(64), nullable=False, server_default="data_access_group_grant", ), sa.Column("feishu_instance_code", sa.String(128), nullable=True), sa.Column("approval_request_id", approval_request_id, nullable=True), sa.Column("idempotency_key", sa.String(128), nullable=True), sa.Column("business_reason", sa.Text(), nullable=True), sa.Column( "status", sa.String(32), nullable=False, server_default="pending_feishu", ), sa.Column("trace_id", sa.String(128), nullable=True), sa.Column( "submitted_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.UniqueConstraint( "feishu_instance_code", name="uq_data_access_requests_feishu_instance_code", ), sa.UniqueConstraint( "idempotency_key", name="uq_data_access_requests_idempotency_key", ), **kw, ) for column, parent, nullable in ( ("requester_person_id", "identity_persons", False), ("principal_email_group_id", "email_groups", False), ("entitlement_id", "data_access_entitlements", False), ("approval_request_id", "approval_requests", True), ): normalize_fk_column_collation( REQUEST_TABLE, column, parent, nullable=nullable, ) op.create_foreign_key( "fk_data_access_requests_requester", REQUEST_TABLE, "identity_persons", ["requester_person_id"], ["id"], ) op.create_foreign_key( "fk_data_access_requests_principal_group", REQUEST_TABLE, "email_groups", ["principal_email_group_id"], ["id"], ) op.create_foreign_key( "fk_data_access_requests_entitlement", REQUEST_TABLE, "data_access_entitlements", ["entitlement_id"], ["id"], ) op.create_foreign_key( "fk_data_access_requests_approval_request", REQUEST_TABLE, "approval_requests", ["approval_request_id"], ["id"], ondelete="SET NULL", ) op.create_index( "ix_data_access_requests_requester", REQUEST_TABLE, ["requester_person_id"], ) op.create_index( "ix_data_access_requests_principal", REQUEST_TABLE, ["principal_email_group_id"], ) op.create_index( "ix_data_access_requests_entitlement", REQUEST_TABLE, ["entitlement_id"], ) op.create_index( "ix_data_access_requests_status", REQUEST_TABLE, ["status"], ) op.create_index( "ix_data_access_requests_approval_request", REQUEST_TABLE, ["approval_request_id"], ) if not _table_exists(bind, ITEM_TABLE): item_id = id_type_for("identity_persons") request_fk = id_type_for(REQUEST_TABLE) resource_id = id_type_for("data_access_resources") person_id = id_type_for("identity_persons") op.create_table( ITEM_TABLE, sa.Column("id", item_id, primary_key=True), sa.Column("request_id", request_fk, nullable=False), sa.Column("resource_id", resource_id, nullable=False), sa.Column("line_no", sa.Integer(), nullable=False), sa.Column( "resource_project_id_snapshot", sa.String(64), nullable=False, ), sa.Column( "resource_dataset_id_snapshot", sa.String(1024), nullable=False, ), sa.Column( "resource_source_version_snapshot", sa.String(128), nullable=False, ), sa.Column( "resource_source_etag_snapshot", sa.String(256), nullable=False, ), sa.Column( "resource_owner_person_id_snapshot", person_id, nullable=True, ), sa.Column( "status", sa.String(32), nullable=False, server_default="pending", ), sa.Column("staleness_reason", sa.String(64), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.UniqueConstraint( "request_id", "line_no", name="uq_data_access_request_items_request_line", ), sa.UniqueConstraint( "request_id", "resource_id", name="uq_data_access_request_items_request_resource", ), **kw, ) for column, parent, nullable in ( ("request_id", REQUEST_TABLE, False), ("resource_id", "data_access_resources", False), ("resource_owner_person_id_snapshot", "identity_persons", True), ): normalize_fk_column_collation( ITEM_TABLE, column, parent, nullable=nullable, ) op.create_foreign_key( "fk_data_access_request_items_request", ITEM_TABLE, REQUEST_TABLE, ["request_id"], ["id"], ondelete="CASCADE", ) op.create_foreign_key( "fk_data_access_request_items_resource", ITEM_TABLE, "data_access_resources", ["resource_id"], ["id"], ) op.create_foreign_key( "fk_data_access_request_items_owner_snapshot", ITEM_TABLE, "identity_persons", ["resource_owner_person_id_snapshot"], ["id"], ondelete="SET NULL", ) op.create_index( "ix_data_access_request_items_resource", ITEM_TABLE, ["resource_id"], ) op.create_index( "ix_data_access_request_items_status", ITEM_TABLE, ["status"], ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, ITEM_TABLE): op.drop_table(ITEM_TABLE) if _table_exists(bind, REQUEST_TABLE): op.drop_table(REQUEST_TABLE)