"""apikey_assignment_members — shared-access roster for an API Key. Revision ID: 200 Revises: 199 Create Date: 2026-05-25 Why this exists --------------- Until now an API Key was a strict 1:1 between the applicant (``ApiKeyAssignment.assigned_to``) and a single human; only that person (plus admins) could read the key's MTD spend / usage detail / cost breakdown. In practice keys are issued for a *team* — the project lead applies, then the colleagues actually using the same SDK credential need to see what's left in the budget. The lead had to either screenshot the spend panel every week or hand over admin to a teammate (both bad). This migration adds a thin shared-access roster: ``Owner`` (= the ``assigned_to`` applicant) maintains a list of teammates ("members") who get read access to the same surfaces as the owner — spend / detail / reveal / list / member-of view. Owner-only operations (member add / remove, transfer, release, revoke, IAM rebind, validity extend) stay exclusively with the Owner. Schema notes ------------ * ``assignment_id`` FK with ``ON DELETE CASCADE``: when a key is hard- deleted (rare; revoke is the norm) the member roster is meaningless and should not survive forensically — the audit log already captures who was on the roster at any past point via ``apikey.member.*`` actions. * ``person_id`` FK with ``ON DELETE CASCADE``: when a person is hard- deleted the membership row carries no useful state — same posture as ``software_seat_allocations.person_id``. Soft-deleted / left employees are handled by the application layer's ``employment_status != 'left'`` filter at read + write time, so the Owner doesn't see a leaver in the roster and a leaver's session can't exercise their old member rights (the runtime leaver gate already blocks mutations; the apikey-domain authz helper additionally checks employment_status on read). * ``added_by`` FK with ``ON DELETE SET NULL``: forensic — "which Owner added this teammate?". Soft-survives when the adder is hard-deleted. * ``UNIQUE(assignment_id, person_id)``: a person is on a key's roster once or not at all. The service's ``add_member`` upserts on this constraint so a double-click from the Owner panel is idempotent. * ``ix_apikey_assignment_members_person_id``: powers the "my member keys" augment on ``GET /apikeys/my`` — every list request runs ``... OR id IN (SELECT assignment_id FROM members WHERE person_id = :me)`` so a member sees the same row their Owner sees. FK column types are mirrored from the target tables at runtime via ``_fk_source_col_type`` (same helper pattern as 192_apikey_release_requests): production ``api_key_assignments.id`` and ``identity_persons.id`` are ``char(36)`` collated ``utf8mb4_unicode_ci``, and a naive ``sa.CHAR(36)`` inherits the connection-time collation which TiDB rejects on FK creation with ``ERROR 3780``. Idempotency ----------- Table creation guarded via ``information_schema.TABLES`` probe (same shape as 191 / 192). Safe to re-run. """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "200" down_revision = "199" branch_labels = None depends_on = None def _table_exists(bind, table: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t LIMIT 1" ), {"t": table}, ).scalar() is not None def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "char", fallback_collation: str | None = "utf8mb4_unicode_ci", length: int = 36, ): """Mirror the target FK column's data type + collation. Same helper migration 192 introduced — see its docstring for the full failure mode (TiDB ERROR 3780 on collation mismatch). """ 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() if _table_exists(bind, "apikey_assignment_members"): return assignment_id_col = _fk_source_col_type( bind, "api_key_assignments", "id", ) person_id_col = _fk_source_col_type( bind, "identity_persons", "id", ) added_by_col = _fk_source_col_type( bind, "identity_persons", "id", ) op.create_table( "apikey_assignment_members", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column( "assignment_id", assignment_id_col, sa.ForeignKey( "api_key_assignments.id", ondelete="CASCADE", ), nullable=False, ), sa.Column( "person_id", person_id_col, sa.ForeignKey( "identity_persons.id", ondelete="CASCADE", ), nullable=False, ), sa.Column( "added_by", added_by_col, sa.ForeignKey( "identity_persons.id", ondelete="SET NULL", ), nullable=True, ), sa.Column( "added_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.UniqueConstraint( "assignment_id", "person_id", name="uq_apikey_assignment_members_assign_person", ), sa.Index( "ix_apikey_assignment_members_person_id", "person_id", ), sa.Index( "ix_apikey_assignment_members_assignment_id", "assignment_id", ), ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "apikey_assignment_members"): op.drop_table("apikey_assignment_members")