"""software one-time add-on charges: items catalog + per-employee charges, plus extend software_monthly_bills(+history) with an addon line type. Lets an employee buy a one-time software add-on (AI credits, vendor person-days, …) through the same request→approval rails as a seat, but WITHOUT becoming a seat holder. Approved charges materialise into ``software_monthly_bills`` as ``line_type='addon'`` rows so every existing cost rollup (finance_spend_coordinator, software-cost views, my-cost) picks the money up via the same SUM(amount) — no parallel table, no lost spend. Revision ID: 348 Revises: 347 """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "348" down_revision = "347" branch_labels = None depends_on = None ITEMS = "software_addon_items" CHARGES = "software_addon_charges" BILLS = "software_monthly_bills" BILL_HISTORY = "software_monthly_bill_history" def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :n" ), {"n": name}, ).scalar() is not None def _column_exists(bind, table: str, column: str) -> bool: return 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}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() # Match the sibling tables created in migration 019/063: addon tables # live in the asset/software collation neighbourhood, person FKs use # the identity_persons id type. Mirroring those exact resolutions # avoids the TiDB 3780 CHAR↔collation FK trap. SW_ID = id_type_for("asset_software_catalog") PERSON_ID = id_type_for("identity_persons") SW_KW = table_kw_for("asset_software_catalog") PERSON_KW = table_kw_for("identity_persons") # 1. software_addon_items — admin-configured add-on definitions # (sibling of software_license_plans). if not _table_exists(bind, ITEMS): op.create_table( ITEMS, sa.Column("id", SW_ID, nullable=False), sa.Column("software_id", SW_ID, nullable=False), sa.Column("name", sa.String(128), nullable=False), sa.Column("unit_label", sa.String(64), nullable=False), sa.Column("unit_price", sa.Numeric(12, 2), nullable=False), sa.Column( "currency", sa.String(3), nullable=False, server_default=sa.text("'CNY'"), ), sa.Column( "min_quantity", sa.Integer, nullable=False, server_default=sa.text("1"), ), sa.Column( "step", sa.Integer, nullable=False, server_default=sa.text("1"), ), sa.Column("max_quantity", sa.Integer, nullable=True), sa.Column("description", sa.Text, nullable=True), sa.Column( "status", sa.String(16), nullable=False, server_default=sa.text("'active'"), ), 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.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint(["software_id"], ["asset_software_catalog.id"]), **SW_KW, ) op.create_index( "ix_software_addon_items_software_id", ITEMS, ["software_id"], ) # 2. software_addon_charges — per-employee one-time charge requests # (sibling of software_seat_allocations: person-collation table). if not _table_exists(bind, CHARGES): op.create_table( CHARGES, sa.Column("id", PERSON_ID, nullable=False), sa.Column("item_id", SW_ID, nullable=False), sa.Column("software_id", SW_ID, nullable=False), sa.Column("user_id", PERSON_ID, nullable=False), sa.Column("quantity", sa.Integer, nullable=False), sa.Column("unit_price", sa.Numeric(12, 2), nullable=False), sa.Column( "currency", sa.String(3), nullable=False, server_default=sa.text("'CNY'"), ), sa.Column("amount", sa.Numeric(12, 2), nullable=False), sa.Column( "status", sa.String(16), nullable=False, server_default=sa.text("'pending'"), ), sa.Column("reason", sa.Text, nullable=True), sa.Column("approval_request_id", PERSON_ID, nullable=True), sa.Column("department", sa.String(128), nullable=True), sa.Column("period", sa.String(7), nullable=True), sa.Column("approved_at", sa.DateTime(timezone=True), nullable=True), sa.Column("rejected_at", sa.DateTime(timezone=True), nullable=True), sa.Column("cancelled_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.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint(["item_id"], [f"{ITEMS}.id"]), sa.ForeignKeyConstraint(["software_id"], ["asset_software_catalog.id"]), sa.ForeignKeyConstraint(["user_id"], ["identity_persons.id"]), **PERSON_KW, ) op.create_index( "ix_software_addon_charges_user_id", CHARGES, ["user_id"], ) op.create_index( "ix_software_addon_charges_software_id", CHARGES, ["software_id"], ) op.create_index( "ix_software_addon_charges_status_period", CHARGES, ["status", "period"], ) op.create_index( "ix_software_addon_charges_period", CHARGES, ["period"], ) op.create_index( "ix_software_addon_charges_item_id", CHARGES, ["item_id"], ) # 3. Extend software_monthly_bills + history with the addon line type. # Additive + backward-compatible: line_type defaults to 'seat' so # every pre-348 row reads correctly without a backfill, and # allocation_id / plan_id are relaxed to NULL for addon lines. for table in (BILLS, BILL_HISTORY): if not _table_exists(bind, table): continue if not _column_exists(bind, table, "line_type"): op.add_column( table, sa.Column( "line_type", sa.String(16), nullable=False, server_default=sa.text("'seat'"), ), ) if not _column_exists(bind, table, "charge_id"): op.add_column(table, sa.Column("charge_id", SW_ID, nullable=True)) if not _column_exists(bind, table, "quantity"): op.add_column(table, sa.Column("quantity", sa.Integer, nullable=True)) # Relax the seat-era NOT NULL so addon lines can leave them blank. op.alter_column( table, "allocation_id", existing_type=SW_ID, nullable=True, ) op.alter_column( table, "plan_id", existing_type=SW_ID, nullable=True, ) def downgrade() -> None: bind = op.get_bind() for table in (BILLS, BILL_HISTORY): if not _table_exists(bind, table): continue for col in ("quantity", "charge_id", "line_type"): if _column_exists(bind, table, col): op.drop_column(table, col) op.drop_table(CHARGES) op.drop_table(ITEMS)