"""Aliyun Bailian (Model Studio) — workspaces / applications / API keys / usage. Revision ID: 074 Revises: 073 Create Date: 2026-04-27 Models the four resources Helix surfaces from a Bailian tenant: * ``bailian_workspaces`` — business workspaces (业务空间) under one Aliyun account; pulled by the ``bailian.workspace_sync`` job. * ``bailian_applications`` — apps / agents / knowledge bases that live inside a workspace; pulled by ``bailian.application_sync``. * ``bailian_api_keys`` — Helix-issued DashScope API keys (their Aliyun-side handle, NOT the secret value — secrets only ever live on the originating ``api_key_assignments.key_value`` row). * ``bailian_usage_daily`` — per-(workspace × api_key × model × day) token + cost rollup, pulled by ``bailian.usage_sync``. All four tables are scoped by ``integration_id`` (the Aliyun cloud integration row that owns the credentials). A future tenant migration to a different Aliyun account is a re-sync, not a schema change. """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "074" down_revision = "073" branch_labels = None depends_on = None def _existing_table_collation(bind, table: str) -> tuple[str, str] | None: """Same charset+collation lookup helper as migration 072. Bailian tables FK to ``integrations.id``, ``identity_persons.id``, and ``api_key_assignments.id``; all three need to share collation with the new tables for the FKs to be accepted. """ row = bind.execute( sa.text( "SELECT t.TABLE_COLLATION, c.CHARACTER_SET_NAME " "FROM information_schema.TABLES t " "JOIN information_schema.COLLATIONS c " " ON c.COLLATION_NAME = t.TABLE_COLLATION " "WHERE t.TABLE_SCHEMA = DATABASE() AND t.TABLE_NAME = :t", ), {"t": table}, ).fetchone() if not row: return None collation, charset = row[0], row[1] if not collation or not charset: return None return charset, collation def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "varchar", fallback_collation: str | None = None, length: int = 36, ): """Return an SA type instance that mirrors an FK target's type+collation. TiDB rejects FKs whose source column's data type differs from the target's — even ``CHAR(36)`` vs ``VARCHAR(36)`` is rejected. MySQL 8.0 is more permissive on the type axis but still strict on collation. The bailian tables FK to three older tables that were created with mixed types: ``integrations.id`` and ``identity_persons.id`` are ``CHAR(36) COLLATE utf8mb4_unicode_ci`` while the bailian tables themselves use ``VARCHAR(36)`` for their PKs. We resolve each target's actual type+collation at runtime so the source column matches the cluster regardless of which historical migration created it. """ row = bind.execute( sa.text( "SELECT DATA_TYPE, COLLATION_NAME FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME = :t AND COLUMN_NAME = :c", ), {"t": target_table, "c": target_column}, ).fetchone() if row: data_type = (row[0] or fallback_type).lower() collation = row[1] or fallback_collation else: data_type = fallback_type.lower() collation = fallback_collation type_cls = sa.CHAR if data_type == "char" else sa.String if collation: return type_cls(length, collation=collation) return type_cls(length) def upgrade() -> None: bind = op.get_bind() parent = ( _existing_table_collation(bind, "integrations") or _existing_table_collation(bind, "identity_persons") or ("utf8mb4", "utf8mb4_0900_ai_ci") ) table_kwargs = {"mysql_charset": parent[0], "mysql_collate": parent[1]} # Resolve the existing-table FK targets' type+collation once. These # are the three tables the bailian schema FKs into; on prod they # are all ``CHAR(36)`` while the bailian PKs we are about to create # are ``VARCHAR(36)`` — so the source columns must mirror the # target type, not the host table type. integrations_id_col = _fk_source_col_type( bind, "integrations", "id", fallback_type="char", fallback_collation=parent[1], ) identity_persons_id_col = _fk_source_col_type( bind, "identity_persons", "id", fallback_type="char", fallback_collation=parent[1], ) api_key_assignments_id_col = _fk_source_col_type( bind, "api_key_assignments", "id", fallback_type="char", fallback_collation=parent[1], ) op.create_table( "bailian_workspaces", sa.Column("id", sa.String(36), primary_key=True), sa.Column( "integration_id", integrations_id_col, sa.ForeignKey("integrations.id", ondelete="CASCADE"), nullable=False, ), sa.Column("workspace_id", sa.String(64), nullable=False), sa.Column("name", sa.String(256), nullable=False), sa.Column("description", sa.Text(), nullable=True), sa.Column("region", sa.String(64), nullable=True), sa.Column("status", sa.String(32), nullable=False, server_default=sa.text("'active'")), sa.Column("is_default", sa.Boolean(), nullable=False, server_default=sa.text("0")), sa.Column( "owner_person_id", identity_persons_id_col, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ), sa.Column("raw_data", sa.JSON(), nullable=True), sa.Column( "first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "last_synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("deleted_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(), server_onupdate=sa.func.now(), ), sa.UniqueConstraint( "integration_id", "workspace_id", name="uq_bailian_workspace_integration_id", ), sa.Index("ix_bailian_workspaces_integration", "integration_id"), **table_kwargs, ) op.create_table( "bailian_applications", sa.Column("id", sa.String(36), primary_key=True), sa.Column( "workspace_pk", sa.String(36), sa.ForeignKey("bailian_workspaces.id", ondelete="CASCADE"), nullable=False, ), sa.Column("application_id", sa.String(128), nullable=False), sa.Column("name", sa.String(256), nullable=False), sa.Column("description", sa.Text(), nullable=True), # 'agent' | 'workflow' | 'rag_app' | 'plugin' | other. sa.Column("application_type", sa.String(32), nullable=True), sa.Column("model_name", sa.String(128), nullable=True), sa.Column("status", sa.String(32), nullable=False, server_default=sa.text("'active'")), sa.Column( "owner_person_id", identity_persons_id_col, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ), sa.Column("raw_data", sa.JSON(), nullable=True), sa.Column( "first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "last_synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("deleted_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(), server_onupdate=sa.func.now(), ), sa.UniqueConstraint( "workspace_pk", "application_id", name="uq_bailian_app_workspace_appid", ), sa.Index("ix_bailian_applications_workspace", "workspace_pk"), **table_kwargs, ) op.create_table( "bailian_api_keys", sa.Column("id", sa.String(36), primary_key=True), sa.Column( "integration_id", integrations_id_col, sa.ForeignKey("integrations.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "workspace_pk", sa.String(36), sa.ForeignKey("bailian_workspaces.id", ondelete="SET NULL"), nullable=True, ), # The Aliyun-side handle (e.g. ak-xxxx). NOT the secret value. sa.Column("api_key_id", sa.String(128), nullable=False), sa.Column("display_name", sa.String(256), nullable=False), # Which Helix person this key was issued to. Always set on # auto-grant; nullable for keys discovered via sync but not yet # claimed. sa.Column( "owner_person_id", identity_persons_id_col, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ), # The RAM user that backs this key (Bailian's permission model # ties API keys to a workspace member, who is a RAM identity). sa.Column("ram_user_name", sa.String(255), nullable=True), # Pointer back to the apikey_assignment if this key was minted # via the Helix request flow. sa.Column( "assignment_id", api_key_assignments_id_col, sa.ForeignKey("api_key_assignments.id", ondelete="SET NULL"), nullable=True, ), sa.Column("status", sa.String(32), nullable=False, server_default=sa.text("'active'")), sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True), sa.Column("raw_data", sa.JSON(), nullable=True), sa.Column( "first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "last_synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("deleted_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(), server_onupdate=sa.func.now(), ), sa.UniqueConstraint( "integration_id", "api_key_id", name="uq_bailian_apikey_integration_apikey", ), sa.Index("ix_bailian_api_keys_owner", "owner_person_id"), sa.Index("ix_bailian_api_keys_workspace", "workspace_pk"), **table_kwargs, ) op.create_table( "bailian_usage_daily", sa.Column("id", sa.String(36), primary_key=True), sa.Column( "integration_id", integrations_id_col, sa.ForeignKey("integrations.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "workspace_pk", sa.String(36), sa.ForeignKey("bailian_workspaces.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "api_key_pk", sa.String(36), sa.ForeignKey("bailian_api_keys.id", ondelete="SET NULL"), nullable=True, ), sa.Column("usage_date", sa.Date(), nullable=False), sa.Column("model_name", sa.String(128), nullable=False), sa.Column( "input_tokens", sa.BigInteger(), nullable=False, server_default=sa.text("0"), ), sa.Column( "output_tokens", sa.BigInteger(), nullable=False, server_default=sa.text("0"), ), sa.Column( "request_count", sa.BigInteger(), nullable=False, server_default=sa.text("0"), ), sa.Column( "cost_usd", sa.Numeric(14, 6), nullable=False, server_default=sa.text("0"), ), sa.Column( "cost_cny", sa.Numeric(14, 6), nullable=False, server_default=sa.text("0"), ), sa.Column("raw_data", sa.JSON(), nullable=True), sa.Column( "synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), 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(), server_onupdate=sa.func.now(), ), sa.UniqueConstraint( "workspace_pk", "api_key_pk", "usage_date", "model_name", name="uq_bailian_usage_day_key_model", ), sa.Index( "ix_bailian_usage_workspace_date", "workspace_pk", "usage_date", ), sa.Index( "ix_bailian_usage_apikey_date", "api_key_pk", "usage_date", ), **table_kwargs, ) def downgrade() -> None: op.drop_table("bailian_usage_daily") op.drop_table("bailian_api_keys") op.drop_table("bailian_applications") op.drop_table("bailian_workspaces")