"""Initial schema. Revision ID: 0001 Revises: Create Date: 2026-07-30 """ from __future__ import annotations import sqlalchemy as sa from alembic import op from sqlalchemy.dialects import postgresql revision = "0001" down_revision = None branch_labels = None depends_on = None # The types are created once, explicitly, below. The column definitions # reference them with create_type=False - without that, create_table tries to # CREATE TYPE again and the migration fails on "type already exists". TICKET_STATUS = postgresql.ENUM( "reported", "triaged", "scheduled", "in_progress", "closed", "void", name="ticket_status", create_type=False, ) TICKET_PRIORITY = postgresql.ENUM( "low", "normal", "high", "critical", name="ticket_priority", create_type=False ) USAGE_DIRECTION = postgresql.ENUM( "used", "returned", name="usage_direction", create_type=False ) def upgrade() -> None: op.execute( "DO $$ BEGIN " "CREATE TYPE ticket_status AS ENUM " "('reported','triaged','scheduled','in_progress','closed','void'); " "EXCEPTION WHEN duplicate_object THEN NULL; END $$" ) op.execute( "DO $$ BEGIN " "CREATE TYPE ticket_priority AS ENUM ('low','normal','high','critical'); " "EXCEPTION WHEN duplicate_object THEN NULL; END $$" ) op.execute( "DO $$ BEGIN " "CREATE TYPE usage_direction AS ENUM ('used','returned'); " "EXCEPTION WHEN duplicate_object THEN NULL; END $$" ) # The auditability guarantee: never reset, never cycled, never reused. # NO CYCLE is the default but is stated explicitly because it is load-bearing. op.execute("CREATE SEQUENCE IF NOT EXISTS ticket_number_seq START WITH 1 NO CYCLE") op.create_table( "assets", sa.Column("id", sa.Integer, primary_key=True), sa.Column("equipment_no", sa.String(64), nullable=False, unique=True), sa.Column("description", sa.String(255), nullable=False), sa.Column("line", sa.String(128)), sa.Column("manufacturer", sa.String(128)), ) op.create_index("ix_assets_equipment_no", "assets", ["equipment_no"]) op.create_index("ix_assets_description", "assets", ["description"]) op.create_table( "parts", sa.Column("id", sa.Integer, primary_key=True), sa.Column("part_number", sa.String(64), nullable=False, unique=True), sa.Column("part_name", sa.String(255), nullable=False), sa.Column("description", sa.String(255)), sa.Column("uom", sa.String(32)), sa.Column("location", sa.String(128)), sa.Column("area", sa.String(128)), sa.Column("part_type", sa.String(64)), sa.Column("barcode", sa.String(128)), sa.Column("unit_cost", sa.Numeric(12, 2)), sa.Column("quantity", sa.Integer, nullable=False, server_default="0"), ) op.create_index("ix_parts_part_number", "parts", ["part_number"]) op.create_index("ix_parts_part_name", "parts", ["part_name"]) op.create_index("ix_parts_barcode", "parts", ["barcode"]) op.create_table( "technicians", sa.Column("id", sa.Integer, primary_key=True), sa.Column("name", sa.String(128), nullable=False, unique=True), sa.Column("role", sa.String(32), nullable=False, server_default="tech"), sa.Column("active", sa.Boolean, nullable=False, server_default=sa.true()), ) op.create_table( "tickets", sa.Column("id", sa.Integer, primary_key=True), sa.Column("number", sa.BigInteger, nullable=False, unique=True), sa.Column("asset_id", sa.Integer, sa.ForeignKey("assets.id")), sa.Column("parent_ticket_id", sa.Integer, sa.ForeignKey("tickets.id")), sa.Column("description", sa.Text, nullable=False), sa.Column("status", TICKET_STATUS, nullable=False, server_default="reported"), sa.Column("priority", TICKET_PRIORITY, nullable=False, server_default="normal"), sa.Column("reported_by_id", sa.Integer, sa.ForeignKey("technicians.id")), sa.Column("assigned_to_id", sa.Integer, sa.ForeignKey("technicians.id")), sa.Column( "reported_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("closed_at", sa.DateTime(timezone=True)), ) op.create_index("ix_tickets_number", "tickets", ["number"]) op.create_table( "ticket_events", sa.Column("id", sa.Integer, primary_key=True), sa.Column( "ticket_id", sa.Integer, sa.ForeignKey("tickets.id"), nullable=False ), sa.Column("kind", sa.String(32), nullable=False), sa.Column("detail", sa.Text), sa.Column("actor_id", sa.Integer, sa.ForeignKey("technicians.id")), sa.Column( "at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), ) op.create_index("ix_ticket_events_ticket_id", "ticket_events", ["ticket_id"]) op.create_table( "ticket_media", sa.Column("id", sa.Integer, primary_key=True), sa.Column( "ticket_id", sa.Integer, sa.ForeignKey("tickets.id"), nullable=False ), sa.Column("rel_path", sa.String(255), nullable=False), sa.Column("mime", sa.String(128), nullable=False), sa.Column("size_bytes", sa.Integer, nullable=False), sa.Column("uploaded_by_id", sa.Integer, sa.ForeignKey("technicians.id")), sa.Column( "uploaded_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), ) op.create_index("ix_ticket_media_ticket_id", "ticket_media", ["ticket_id"]) op.create_table( "part_usage", sa.Column("id", sa.Integer, primary_key=True), sa.Column( "ticket_id", sa.Integer, sa.ForeignKey("tickets.id"), nullable=False ), sa.Column("part_id", sa.Integer, sa.ForeignKey("parts.id"), nullable=False), sa.Column("quantity", sa.Numeric(12, 3), nullable=False), sa.Column( "direction", USAGE_DIRECTION, nullable=False, server_default="used" ), sa.Column("actor_id", sa.Integer, sa.ForeignKey("technicians.id")), sa.Column( "at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), ) op.create_index("ix_part_usage_ticket_id", "part_usage", ["ticket_id"]) op.create_index("ix_part_usage_part_id", "part_usage", ["part_id"]) def downgrade() -> None: op.drop_table("part_usage") op.drop_table("ticket_media") op.drop_table("ticket_events") op.drop_table("tickets") op.drop_table("technicians") op.drop_table("parts") op.drop_table("assets") op.execute("DROP SEQUENCE IF EXISTS ticket_number_seq") op.execute("DROP TYPE IF EXISTS usage_direction") op.execute("DROP TYPE IF EXISTS ticket_priority") op.execute("DROP TYPE IF EXISTS ticket_status")