"""Asset holder_kind + holder_office_id (office-held first-class mode). Revision ID: 186 Revises: 183 Create Date: 2026-05-19 Renumbered from 183 → 186 during merge — PR #403's ``183_update_laptop_replacement_approval_code.py`` landed first and claimed the 183 slot. Chain order unchanged (still appends schema for office-held mode after the laptop-replacement approval seed); body identical to the original 183. Why this exists --------------- Every Helix asset historically had to be attributed to a ``Person`` via ``assigned_to`` — even devices that physically live in a meeting room or on a shared dock at a city office. The inventory routing in ``inventory_service._classify_asset`` already paper-mached this gap by matching ``location`` strings against ``asset_office_addresses.city`` to find an office admin recipient. That works for ``assigned_to IS NULL`` rows but breaks the moment the T+0 leaver force-transfer puts a leaver's laptop on some IT admin's head — now the asset is ``assigned`` to an active person, the office-admin routing never fires, and 盘点 fan-out piles onto whoever inherited the leaver assets. This migration introduces ``holder_kind`` as an explicit first-class mode on every ``asset_items`` row + a strong FK ``holder_office_id`` pointing at ``asset_office_addresses``. Combined with the new ``held_by_office`` status (free string, no schema change needed — ``status`` is ``VARCHAR(32)``), the asset model now distinguishes: - ``holder_kind='personal'`` — has (or will have) a Person assignee via ``assigned_to``. Default for every existing row; default for new intake when no office is chosen. - ``holder_kind='office'`` — lives at a physical office; routed to that office's ``admin_person_id`` for inventory + offboarding handoffs. ``assigned_to`` MUST be NULL; ``holder_office_id`` points at the responsible office. The invariant ``holder_kind='office' → assigned_to IS NULL`` is enforced at the application layer (service-layer validators) rather than DB CHECK because TiDB 8.0 CHECK constraint semantics are still quirky around partial-row UPDATE paths and we already rely on service-layer guards for every other AssetItem invariant (status transitions, borrow restoration, etc.). FK column type -------------- ``holder_office_id`` mirrors ``asset_office_addresses.id`` — ``CHAR(36)`` with the same collation. We resolve the target column's ``DATA_TYPE`` via INFORMATION_SCHEMA at runtime (mirroring migration 160 / 173) so TiDB never rejects the FK with error 3780 (CHAR vs VARCHAR / collation drift). Idempotency ----------- Every column / index / FK is guarded by an existence check. A partial apply (column landed but FK creation tripped 3780, e.g.) re-runs cleanly. The data backfill that maps existing rows to ``holder_kind='office'`` where appropriate lives in migration 185, NOT here — keeping schema DDL and data fixes in separate migrations is the convention every prior multi-step refactor in this codebase has followed (cf. 042 → 049 split for software catalog owner FKs). """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "186" down_revision = "183" branch_labels = None depends_on = None # ─── helpers (mirror migration 173) ──────────────────────────────── def _fk_source_col_type( bind, target_table: str, target_column: str, *, length: int = 36 ): 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() data_type = (row[0] or "char").lower() if row else "char" collation = row[1] if row else None type_cls = sa.CHAR if data_type == "char" else sa.String return type_cls(length, collation=collation) if collation else type_cls(length) def _column_exists(bind, table: str, column: str) -> bool: row = bind.execute( sa.text( "SELECT 1 FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND COLUMN_NAME = :c" ), {"t": table, "c": column}, ).fetchone() return row is not None def _index_exists(bind, table: str, index: str) -> bool: row = bind.execute( sa.text( "SELECT 1 FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND INDEX_NAME = :i" ), {"t": table, "i": index}, ).fetchone() return row is not None def _fk_exists(bind, table: str, constraint: str) -> bool: row = bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLE_CONSTRAINTS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND CONSTRAINT_NAME = :c " " AND CONSTRAINT_TYPE = 'FOREIGN KEY'" ), {"t": table, "c": constraint}, ).fetchone() return row is not None # ─── upgrade ─────────────────────────────────────────────────────── def upgrade() -> None: bind = op.get_bind() office_id_type = _fk_source_col_type(bind, "asset_office_addresses", "id") # 1) holder_kind: VARCHAR(16) NOT NULL DEFAULT 'personal'. if not _column_exists(bind, "asset_items", "holder_kind"): op.add_column( "asset_items", sa.Column( "holder_kind", sa.String(16), nullable=False, server_default="personal", ), ) # 2) holder_office_id: nullable CHAR(36) FK -> asset_office_addresses.id. if not _column_exists(bind, "asset_items", "holder_office_id"): op.add_column( "asset_items", sa.Column("holder_office_id", office_id_type, nullable=True), ) else: # Repair type drift from a partial apply. op.alter_column( "asset_items", "holder_office_id", type_=office_id_type, existing_nullable=True, ) # 3) Composite index for inventory office-routing + holder-kind filters. if not _index_exists(bind, "asset_items", "ix_asset_items_holder"): op.create_index( "ix_asset_items_holder", "asset_items", ["holder_kind", "holder_office_id"], ) # 4) FK to asset_office_addresses.id. SET NULL on delete — an office # row can be archived/removed, but we keep the asset row alive # and surface a UI warning (no holder_office_id → falls back to # location / IT fallback at classify time). fk_name = "fk_asset_items_holder_office_id" if not _fk_exists(bind, "asset_items", fk_name): op.create_foreign_key( fk_name, "asset_items", "asset_office_addresses", ["holder_office_id"], ["id"], ondelete="SET NULL", ) # ─── downgrade ───────────────────────────────────────────────────── def downgrade() -> None: bind = op.get_bind() fk_name = "fk_asset_items_holder_office_id" if _fk_exists(bind, "asset_items", fk_name): op.drop_constraint(fk_name, "asset_items", type_="foreignkey") if _index_exists(bind, "asset_items", "ix_asset_items_holder"): op.drop_index("ix_asset_items_holder", table_name="asset_items") if _column_exists(bind, "asset_items", "holder_office_id"): op.drop_column("asset_items", "holder_office_id") if _column_exists(bind, "asset_items", "holder_kind"): op.drop_column("asset_items", "holder_kind")