Database Migrations

MUST READ before creating any Alembic migration.

This guide defines the exact conventions for creating database migrations in Hegemony. AI agents and developers must follow these patterns precisely.

Quick Reference

Aspect	Convention
File naming	`YYYYMMDD_NNN_description.py` (e.g., `20260204_024_add_jobs.py`)
Revision ID	`NNN_short_description` (e.g., `024_add_jobs`)
down_revision	Previous migration's revision ID (e.g., `023_remove_timezone`)
Location	`apps/api/alembic/versions/`

Baseline: the migration chain was squashed on 2026-05-12 to a single revision 001_initial (file 20260512_001_initial.py). The previous 60 revisions were removed and existing databases are not upgradable from them. Numbering restarted at 002_* after the squash; find the current head as described below.

Before Creating a Migration

1. Find the Latest Migration

bash

ls -la apps/api/alembic/versions/ | tail -5

Note the highest NNN number and its revision ID. Your new migration:

Uses NNN + 1 for the sequence number
Sets down_revision to the previous migration's revision ID

2. Verify the Revision Chain

bash

task db:history

Ensure no gaps or duplicates in the chain.

Migration File Template

Copy this exact structure. Replace placeholders in <angle_brackets>.

python

"""<Short description of what this migration does>.

Revision ID: <NNN>_<short_description>
Revises: <previous_revision_id>
Create Date: <YYYY-MM-DD HH:MM:SS.ffffff>

Adds:
- <table/column/index 1>
- <table/column/index 2>
"""

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision: str = "<NNN>_<short_description>"
down_revision: str | None = "<previous_revision_id>"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
    # === Tables ===

    # === Columns ===

    # === Indexes ===

    # === Constraints ===
    pass


def downgrade() -> None:
    # Reverse order of upgrade operations
    pass

Naming Conventions

File Name

text

YYYYMMDD_NNN_description.py
│        │   └── Snake_case, max 40 chars
│        └── Three-digit sequence, zero-padded (001, 024, 100)
└── Current date in UTC

Naming guidance:

Omit add_ for new tables: scheduled_jobs not add_scheduled_jobs
Use add_ when adding columns/enums to existing tables: add_device_status_enum
Use verbs for other operations: rename_foo_to_bar, drop_legacy_columns

Examples:

✅ 20260204_024_scheduled_jobs.py (new table)
✅ 20260204_025_add_device_status_enum.py (adding enum to existing table)
❌ 024_scheduled_jobs.py (missing date)
❌ 20260204_24_jobs.py (not zero-padded)

Revision ID

text

NNN_short_description
│   └── Matches file description, underscores
└── Same three-digit number as filename

Examples:

✅ 024_scheduled_jobs
✅ 025_add_device_status_enum
❌ 20260204_024_scheduled_jobs (don't include date in revision ID)
❌ scheduled_jobs (missing sequence number)

Common Patterns

Creating a Table

python

def upgrade() -> None:
    op.create_table(
        "scheduled_jobs",
        sa.Column("id", sa.UUID(), primary_key=True),
        sa.Column("name", sa.String(255), nullable=False),
        sa.Column("flow_id", sa.UUID(), sa.ForeignKey("flows.id", ondelete="CASCADE"), nullable=False),
        sa.Column("cron_expression", sa.String(100), nullable=False),
        sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.text("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()),
    )
    # Always create indexes for foreign keys
    op.create_index("ix_scheduled_jobs_flow_id", "scheduled_jobs", ["flow_id"])


def downgrade() -> None:
    op.drop_index("ix_scheduled_jobs_flow_id", table_name="scheduled_jobs")
    op.drop_table("scheduled_jobs")

Adding a Column

python

def upgrade() -> None:
    op.add_column(
        "flows",
        sa.Column("schedule_id", sa.UUID(), sa.ForeignKey("scheduled_jobs.id"), nullable=True),
    )


def downgrade() -> None:
    op.drop_column("flows", "schedule_id")

Creating an Enum Type (PostgreSQL)

python

def upgrade() -> None:
    # Use EXECUTE for enum creation - allows IF NOT EXISTS
    op.execute("""
        DO $$
        BEGIN
            IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'jobstatus') THEN
                CREATE TYPE jobstatus AS ENUM ('pending', 'running', 'completed', 'failed');
            END IF;
        END
        $$;
    """)

    op.add_column(
        "scheduled_jobs",
        sa.Column("status", sa.Enum("pending", "running", "completed", "failed", name="jobstatus"), nullable=False),
    )


def downgrade() -> None:
    op.drop_column("scheduled_jobs", "status")
    op.execute("DROP TYPE IF EXISTS jobstatus")

Caveat — enums declared inline in op.create_table. When a new table's column uses sa.Enum("a", "b", name="foo"), Postgres auto-creates the foo type as a side effect of the CREATE TABLE, but op.drop_table does not drop it. Downgrade must explicitly op.execute("DROP TYPE IF EXISTS foo CASCADE") for every such enum, or a subsequent upgrade head will fail with DuplicateObjectError: type "foo" already exists.

Adding Values to Existing Enum

python

def upgrade() -> None:
    # PostgreSQL: Add enum value (cannot be in transaction, use raw connection)
    op.execute("ALTER TYPE runstatus ADD VALUE IF NOT EXISTS 'scheduled'")


def downgrade() -> None:
    # Note: PostgreSQL doesn't support removing enum values easily
    # Document this limitation in the docstring
    pass

Creating Indexes

python

def upgrade() -> None:
    # Naming convention: ix_<table>_<column(s)>
    op.create_index("ix_runs_flow_id", "runs", ["flow_id"])
    op.create_index("ix_runs_status_created", "runs", ["status", "created_at"])  # Composite
    op.create_index("ix_runs_metadata", "runs", ["metadata"], postgresql_using="gin")  # GIN for JSONB


def downgrade() -> None:
    op.drop_index("ix_runs_metadata", table_name="runs")
    op.drop_index("ix_runs_status_created", table_name="runs")
    op.drop_index("ix_runs_flow_id", table_name="runs")

Foreign Key Constraints

python

def upgrade() -> None:
    # Naming convention: fk_<table>_<column>_<referenced_table>
    op.create_foreign_key(
        "fk_runs_schedule_id_scheduled_jobs",
        "runs",
        "scheduled_jobs",
        ["schedule_id"],
        ["id"],
        ondelete="SET NULL",
    )


def downgrade() -> None:
    op.drop_constraint("fk_runs_schedule_id_scheduled_jobs", "runs", type_="foreignkey")

Circular Foreign Keys (`use_alter=True`)

When two tables reference each other (e.g. platform_sync_profiles.last_run_id → platform_sync_runs.id and platform_sync_runs.profile_id → platform_sync_profiles.id), the cycle prevents SQLAlchemy from topologically sorting Base.metadata.sorted_tables and Alembic falls back to alphabetical ordering, producing migrations whose CREATE TABLE statements fail.

ORM side — declare the FK on the "breaking" side with use_alter=True and a stable name:

python

last_run_id: Mapped[uuid.UUID | None] = mapped_column(
    UUID(as_uuid=True),
    ForeignKey(
        "platform_sync_runs.id",
        ondelete="SET NULL",
        use_alter=True,
        name="fk_platform_sync_profiles_last_run_id",
    ),
    nullable=True,
)

Verify the cycle is broken before regenerating autogen:

python

python -c "from apps.api import models; from apps.api.db import Base; \
  print([t.name for t in Base.metadata.sorted_tables])"

No SAWarning: Cannot correctly sort tables should appear.

Migration side — autogenerate inlines the ForeignKeyConstraint(..., use_alter=True) into op.create_table, but Postgres silently drops it because the referenced table does not yet exist. You must add an explicit op.create_foreign_key after both tables are created, and a matching op.drop_constraint at the top of downgrade() (before any drop_table):

python

def upgrade() -> None:
    op.create_table("platform_sync_profiles", ...)  # FK omitted by Postgres
    op.create_table("platform_sync_runs", ...)
    op.create_foreign_key(
        "fk_platform_sync_profiles_last_run_id",
        "platform_sync_profiles",
        "platform_sync_runs",
        ["last_run_id"],
        ["id"],
        ondelete="SET NULL",
    )


def downgrade() -> None:
    op.drop_constraint(
        "fk_platform_sync_profiles_last_run_id",
        "platform_sync_profiles",
        type_="foreignkey",
    )
    op.drop_table("platform_sync_runs")
    op.drop_table("platform_sync_profiles")

Always confirm with SELECT count(*) FROM pg_constraint WHERE contype='f' before and after upgrade that the count matches the number of FKs in the ORM.

Pre-Flight Checklist

Before committing a migration, verify:

[ ] File name follows YYYYMMDD_NNN_description.py
[ ] Revision ID follows NNN_short_description (matches file)
[ ] down_revision points to the correct previous migration
[ ] Type annotations present: revision: str = ..., down_revision: str | None = ...
[ ] Docstring includes Revision ID, Revises, Create Date
[ ] upgrade() and downgrade() are both implemented
[ ] downgrade() reverses operations in correct order
[ ] Indexes created for all foreign key columns
[ ] Naming conventions followed for indexes (ix_), constraints (fk_, uq_)
[ ] Audit rows untouched, or the migration runs SET LOCAL hegemony.audit_maintenance = 'on' first: audit_logs is append-only at the database level (migration 048), and any UPDATE or DELETE on it is refused otherwise

Local Verification Commands

bash

# Apply migration
task db:migrate

# Verify it applied
task db:current

# Test downgrade (CAUTION: destroys data)
task db:downgrade

# Re-apply
task db:migrate

Troubleshooting

"Target database is not up to date"

The database has pending migrations. Run:

bash

task db:migrate

"Can't locate revision"

Your down_revision points to a non-existent revision. Check:

bash

task db:history

"Revision already exists"

Duplicate revision ID. Change your NNN to the next available number.

"value too long for type character varying(32)" when stamping a revision

Older databases created alembic_version.version_num as VARCHAR(32), so a revision ID longer than 32 characters failed when Alembic recorded the new head. Migration 013_widen_alembic_version widens this column to VARCHAR(255), so revision IDs can now be descriptive. If you see this error, run task db:migrate to apply migration 013 (or later).

Multiple heads (branched history)

Two migrations share the same down_revision. They must be merged:

bash

task db:heads  # Lists current head revision(s); multiple entries indicate a branch

Use task db:heads to inspect current heads, create the merge revision using your standard migration workflow, then run task db:heads again to confirm a single head.

Then apply the resulting merge migration:

bash

task db:migrate

What NOT to Do

❌ Don't	✅ Do Instead
Use timestamps in revision ID	Use `NNN_description` format
Skip downgrade implementation	Always implement downgrade
Hardcode UUIDs in migrations	Use `sa.text("gen_random_uuid()")` for defaults
Create migration without testing locally	Always run `task db:migrate` first
Use `op.execute()` for simple operations	Use typed Alembic operations
Forget indexes on foreign keys	Create index for every FK column

Database Migrations ​

Quick Reference ​

Before Creating a Migration ​

1. Find the Latest Migration ​

2. Verify the Revision Chain ​

Migration File Template ​

Naming Conventions ​

File Name ​

Revision ID ​

Common Patterns ​

Creating a Table ​

Adding a Column ​

Creating an Enum Type (PostgreSQL) ​

Adding Values to Existing Enum ​

Creating Indexes ​

Foreign Key Constraints ​

Circular Foreign Keys (use_alter=True) ​

Pre-Flight Checklist ​

Local Verification Commands ​

Troubleshooting ​

"Target database is not up to date" ​

"Can't locate revision" ​

"Revision already exists" ​

"value too long for type character varying(32)" when stamping a revision ​

Multiple heads (branched history) ​

What NOT to Do ​

Database Migrations

Quick Reference

Before Creating a Migration

1. Find the Latest Migration

2. Verify the Revision Chain

Migration File Template

Naming Conventions

File Name

Revision ID

Common Patterns

Creating a Table

Adding a Column

Creating an Enum Type (PostgreSQL)

Adding Values to Existing Enum

Creating Indexes

Foreign Key Constraints

Circular Foreign Keys (`use_alter=True`)

Pre-Flight Checklist

Local Verification Commands

Troubleshooting

"Target database is not up to date"

"Can't locate revision"

"Revision already exists"

"value too long for type character varying(32)" when stamping a revision

Multiple heads (branched history)

What NOT to Do