Beginner

# Run from the project root directory
alembic init alembic
# => creates alembic.ini in the current directory (project config file)
# => creates alembic/ subdirectory (migration environment)
# => creates alembic/env.py (Python script that runs during migrations)
# => creates alembic/script.py.mako (Mako template for new revision files)
# => creates alembic/versions/ (empty directory; revision files go here)
 
# Output printed to terminal:
# => Creating directory /path/to/project/alembic ...  done
# => Generating /path/to/project/alembic.ini ...  done
# => Generating /path/to/project/alembic/env.py ...  done
# => Generating /path/to/project/alembic/script.py.mako ...  done
# => Generating /path/to/project/alembic/versions/.keep ...  done
# => Please edit configuration/connection/logging settings
# =>   in '/path/to/project/alembic.ini' before proceeding.
 
# The argument "alembic" is the name of the generated directory.
# Convention: use "alembic" as the directory name; some teams use "migrations".
# After init, edit alembic.ini to set sqlalchemy.url before running any migration.

# alembic.ini — project root configuration file
 
[alembic]
# script_location: path to the alembic directory (from project root)
script_location = alembic
# => tells Alembic where to find env.py, versions/, and script.py.mako
# => use relative path from project root; "alembic" is the conventional name
 
# file_template: naming pattern for generated revision files
file_template = %%(rev)s_%%(slug)s
# => %%(rev)s  => revision ID hash (e.g., "abc123de")
# => %%(slug)s => slugified version of the -m message (e.g., "create_users_table")
# => result: "abc123de_create_users_table.py"
# => alternative: %%(year)d%%(month).2d%%(day).2d_%%(rev)s_%%(slug)s (date prefix)
 
# prepend_sys_path: directories to add to sys.path when env.py runs
prepend_sys_path = .
# => "." means project root is on sys.path
# => enables "from myapp.models import Base" in env.py without install
 
# version_path_separator: OS-appropriate separator for version_locations
version_path_separator = os
# => "os" uses os.pathsep (: on Unix, ; on Windows)
# => only relevant for multi-directory version setups
 
# sqlalchemy.url: database connection string
# sqlalchemy.url = postgresql+psycopg2://user:pass@localhost/dbname
# => LEFT BLANK HERE: set programmatically in env.py from DATABASE_URL env var
# => never hard-code credentials in ini files; use environment variables
 
[loggers]
keys = root,sqlalchemy,alembic
# => three logger namespaces: root (all), sqlalchemy.engine, alembic
 
[handlers]
keys = console
# => single handler: writes to stderr
 
[formatters]
keys = generic
 
[logger_root]
level = WARN
# => suppress INFO and DEBUG from all libraries
handlers = console
qualname =
 
[logger_sqlalchemy]
level = WARN
# => set to INFO to log all SQL statements during migrations (useful for debugging)
handlers =
qualname = sqlalchemy.engine
 
[logger_alembic]
level = INFO
# => INFO shows migration progress; set to DEBUG for verbose output
handlers =
qualname = alembic
 
[handler_console]
class = StreamHandler
args = (sys.stderr,)
# => writes to stderr, not stdout; keeps migration output separate from app output
level = NOTSET
formatter = generic
 
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
# => example output: "INFO  [alembic.runtime.migration] Running upgrade  -> abc123, create users"

# alembic/env.py — full file content
 
"""Alembic environment configuration."""
 
import os
import sys
from logging.config import fileConfig
from pathlib import Path
 
from alembic import context
from sqlalchemy import engine_from_config, pool
 
# --- sys.path setup ---
# Add the src directory so "from myapp.models import Base" works.
_src_path = Path(__file__).parents[2] / "src"
# => Path(__file__) is the path to this env.py file
# => .parents[2] goes two levels up: alembic/ -> project_root/
# => / "src" appends the src directory
if str(_src_path) not in sys.path:
    sys.path.insert(0, str(_src_path))
    # => prepend so project models take precedence over installed packages
 
# --- Import SQLAlchemy metadata ---
from myapp.infrastructure.models import Base  # noqa: E402
# => Base.metadata holds all Table definitions from SQLAlchemy ORM models
# => used by autogenerate to detect schema differences
# => noqa: E402 suppresses "module level import not at top of file" lint warning
 
# --- Alembic config object ---
config = context.config
# => context.config wraps alembic.ini; provides get_main_option(), set_main_option()
 
# --- Logging setup ---
if config.config_file_name is not None:
    fileConfig(config.config_file_name)
    # => applies [loggers] / [handlers] / [formatters] from alembic.ini
 
# --- Autogenerate target ---
target_metadata = Base.metadata
# => must be set for --autogenerate to work
# => None here means autogenerate cannot detect model changes
 
# --- Database URL from environment ---
_database_url = os.environ.get("DATABASE_URL") or config.get_main_option("sqlalchemy.url")
# => prefer DATABASE_URL env var; fall back to alembic.ini value
# => this keeps credentials out of version control
if _database_url:
    config.set_main_option("sqlalchemy.url", _database_url)
    # => overrides whatever was in alembic.ini
 
 
def run_migrations_offline() -> None:
    """Run migrations in offline mode (emit SQL without a live connection)."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,                          # => pass URL string directly (no engine)
        target_metadata=target_metadata,  # => needed for autogenerate in offline mode
        literal_binds=True,               # => render bind parameters as literals in SQL
        dialect_opts={"paramstyle": "named"},
        # => "named" uses :param notation; "qmark" uses ? notation
    )
    with context.begin_transaction():
        context.run_migrations()
        # => writes SQL to stdout or a file; no actual DB connection made
 
 
def run_migrations_online() -> None:
    """Run migrations in online mode (with a live database connection)."""
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        # => reads [alembic] section from alembic.ini for sqlalchemy.* keys
        prefix="sqlalchemy.",             # => strips "sqlalchemy." prefix when passing to create_engine
        poolclass=pool.NullPool,          # => NullPool: no connection reuse; safe for migration scripts
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,        # => live connection object
            target_metadata=target_metadata,
        )
        with context.begin_transaction():
            context.run_migrations()
            # => executes upgrade() or downgrade() from each revision file
            # => updates alembic_version table after each revision
 
 
if context.is_offline_mode():
    run_migrations_offline()
    # => triggered by: alembic upgrade head --sql
else:
    run_migrations_online()
    # => triggered by: alembic upgrade head (no --sql flag)

# Create a revision with no auto-generated content (blank template)
alembic revision -m "create users table"
# => -m "create users table": sets the human-readable message; becomes the slug in the filename
# => generates: alembic/versions/abc123de_create_users_table.py
# => revision ID "abc123de" is a random 8-character hex string
# => the file contains empty upgrade() and downgrade() functions to fill in
 
# Output:
# => Generating /path/to/project/alembic/versions/abc123de_create_users_table.py ...  done
 
# Without -m flag the file is still created but has an empty slug:
alembic revision
# => generates: alembic/versions/def456gh_.py
# => avoid this; always use -m for readable filenames

# alembic/versions/abc123de_create_users_table.py
 
"""create users table
 
Revision ID: abc123de
Revises:
Create Date: 2026-03-27 10:00:00.000000
 
"""
 
from collections.abc import Sequence
 
from alembic import op
import sqlalchemy as sa
 
# revision identifiers, used by Alembic.
revision: str = "abc123de"
# => this revision's unique identifier; referenced by other revisions as down_revision
 
down_revision: str | None = None
# => None means this is the FIRST migration (no parent)
# => for subsequent revisions this will be the parent's revision ID (e.g., "abc123de")
 
branch_labels: str | Sequence[str] | None = None
# => used for named branch management (advanced); None for linear history
 
depends_on: str | Sequence[str] | None = None
# => used for cross-branch dependencies (advanced); None for standard use
 
 
def upgrade() -> None:
    # => called by: alembic upgrade head (or alembic upgrade abc123de)
    # => fill in DDL operations here using op.*
    pass
 
 
def downgrade() -> None:
    # => called by: alembic downgrade -1 (or alembic downgrade base)
    # => fill in REVERSE operations here; must exactly undo upgrade()
    pass

# alembic/versions/abc123de_create_users_table.py
 
"""create users table
 
Revision ID: abc123de
Revises:
Create Date: 2026-03-27 10:00:00.000000
 
"""
 
from alembic import op
import sqlalchemy as sa
 
revision: str = "abc123de"
down_revision: str | None = None
branch_labels = None
depends_on = None
 
 
def upgrade() -> None:
    # => upgrade() is called when running: alembic upgrade head (or upgrade abc123de)
    # => must contain all DDL to move the database FORWARD to this revision
    # => Alembic wraps this in a transaction; if it raises, the transaction rolls back
 
    op.create_table(                          # => emits: CREATE TABLE users (...)
        "users",
        sa.Column("id", sa.Integer, primary_key=True),
        # => id INTEGER PRIMARY KEY
        sa.Column("username", sa.String(50), nullable=False),
        # => username VARCHAR(50) NOT NULL
        sa.Column("email", sa.String(255), nullable=False),
        # => email VARCHAR(255) NOT NULL
    )
    # => after upgrade() returns: alembic_version table is updated to "abc123de"
 
 
def downgrade() -> None:
    # => downgrade() is called when running: alembic downgrade -1 (or downgrade base)
    # => must EXACTLY reverse everything in upgrade()
    # => order matters: reverse the upgrade() operations in reverse order
 
    op.drop_table("users")
    # => emits: DROP TABLE users
    # => after downgrade() returns: alembic_version row for "abc123de" is deleted
    # => the database is now at the previous revision (or no revision if base)

# Inside upgrade() of a revision file
 
from alembic import op
import sqlalchemy as sa
 
def upgrade() -> None:
    op.create_table(
        "products",
        # => first positional argument: table name string
        # => remaining arguments: Column objects defining the schema
 
        sa.Column("id", sa.Integer, primary_key=True),
        # => id INTEGER PRIMARY KEY
        # => primary_key=True: Alembic adds PRIMARY KEY constraint
        # => for PostgreSQL with SERIAL, use autoincrement=True (default for Integer PK)
 
        sa.Column("name", sa.String(200), nullable=False),
        # => name VARCHAR(200) NOT NULL
 
        sa.Column("price", sa.Numeric(10, 2), nullable=False),
        # => price NUMERIC(10,2) NOT NULL
        # => Numeric(precision, scale): 10 total digits, 2 after decimal
 
        sa.Column("description", sa.Text, nullable=True),
        # => description TEXT (nullable, so no NOT NULL constraint)
        # => nullable=True is the default; shown here explicitly for clarity
 
        sa.Column("stock", sa.Integer, nullable=False, server_default="0"),
        # => stock INTEGER NOT NULL DEFAULT 0
        # => server_default="0": the database applies this default, not Python
        # => use server_default for column defaults in CREATE TABLE
 
        sa.Column("active", sa.Boolean, nullable=False, server_default=sa.true()),
        # => active BOOLEAN NOT NULL DEFAULT TRUE
        # => sa.true() is the dialect-correct way to specify TRUE literal
 
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("NOW()"),
        ),
        # => created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
        # => sa.text("NOW()") wraps a raw SQL expression as the default
        # => timezone=True: TIMESTAMPTZ (with time zone) in PostgreSQL
    )
    # => emits the full CREATE TABLE products (...) statement
    # => the table is created in the current schema (usually "public" in PostgreSQL)

# Inside upgrade() — adding a column to an existing "products" table
 
from alembic import op
import sqlalchemy as sa
 
def upgrade() -> None:
    op.add_column(
        "products",
        # => first argument: table name
        sa.Column("sku", sa.String(50), nullable=True),
        # => second argument: Column object defining the new column
        # => sku VARCHAR(50) (nullable because existing rows have no value yet)
        # => nullable=True is necessary for adding a column to a populated table
        # => if nullable=False without a server_default, PostgreSQL raises an error
        # => for NOT NULL columns on existing tables, see Example 21
    )
    # => emits: ALTER TABLE products ADD COLUMN sku VARCHAR(50)
 
 
def downgrade() -> None:
    op.drop_column("products", "sku")
    # => emits: ALTER TABLE products DROP COLUMN sku
    # => reverses the add_column exactly
    # => WARNING: this permanently deletes the column and all its data

# Inside upgrade() — removing a deprecated column from "products"
 
from alembic import op
import sqlalchemy as sa
 
def upgrade() -> None:
    op.drop_column("products", "legacy_code")
    # => first argument: table name
    # => second argument: column name to drop
    # => emits: ALTER TABLE products DROP COLUMN legacy_code
    # => all data in legacy_code is permanently deleted
    # => PostgreSQL also drops any indexes or constraints on this column
 
 
def downgrade() -> None:
    op.add_column(
        "products",
        sa.Column("legacy_code", sa.String(20), nullable=True),
        # => re-create the column on rollback
        # => data is LOST — downgrade re-creates the column structure but not the values
        # => nullable=True: on rollback, existing rows will have NULL in this column
    )
    # => emits: ALTER TABLE products ADD COLUMN legacy_code VARCHAR(20)
    # => WARNING: data is not restored; downgrade only restores schema structure

# Inside upgrade() — adding indexes to the "products" table
 
from alembic import op
 
def upgrade() -> None:
    # Simple single-column index
    op.create_index(
        "ix_products_sku",     # => index name: convention is "ix_{table}_{column}"
        "products",            # => table name
        ["sku"],               # => list of column names to index
        unique=False,          # => non-unique index; allows duplicate SKU values
    )
    # => emits: CREATE INDEX ix_products_sku ON products (sku)
 
    # Unique index enforces uniqueness at the database level
    op.create_index(
        "ix_products_sku_unique",
        "products",
        ["sku"],
        unique=True,           # => unique=True: adds UNIQUE constraint via index
    )
    # => emits: CREATE UNIQUE INDEX ix_products_sku_unique ON products (sku)
    # => uniqueness is enforced by the database; duplicate inserts raise IntegrityError
 
 
def downgrade() -> None:
    op.drop_index("ix_products_sku_unique", table_name="products")
    # => emits: DROP INDEX ix_products_sku_unique
    # => table_name is required so Alembic knows the schema context
    op.drop_index("ix_products_sku", table_name="products")
    # => reverse order: drop unique index first, then non-unique

# Inside upgrade() — adding a foreign key from "orders" to "users"
 
from alembic import op
 
def upgrade() -> None:
    op.create_foreign_key(
        "fk_orders_user_id",   # => constraint name: convention is "fk_{table}_{column}"
        "orders",              # => source table (child) — the table with the FK column
        "users",               # => referent table (parent) — the table being referenced
        ["user_id"],           # => source columns: the FK column(s) in "orders"
        ["id"],                # => referent columns: the PK column(s) in "users"
        ondelete="RESTRICT",   # => RESTRICT: prevent deleting a user who has orders
        # => other options: "CASCADE" (delete orders when user deleted)
        # =>                "SET NULL" (set user_id to NULL when user deleted)
        # =>                "SET DEFAULT" (set to column default)
        # =>                "NO ACTION" (default; deferred check, similar to RESTRICT)
    )
    # => emits: ALTER TABLE orders ADD CONSTRAINT fk_orders_user_id
    # =>        FOREIGN KEY (user_id) REFERENCES users (id) ON DELETE RESTRICT
 
 
def downgrade() -> None:
    op.drop_constraint("fk_orders_user_id", "orders", type_="foreignkey")
    # => first argument: constraint name
    # => second argument: table name where the constraint lives
    # => type_="foreignkey": required to identify constraint type
    # => emits: ALTER TABLE orders DROP CONSTRAINT fk_orders_user_id

# Apply all pending migrations up to the latest (head) revision
alembic upgrade head
# => reads alembic.ini to find script_location and database URL
# => connects to the database
# => checks alembic_version table for the current revision
# => runs upgrade() for each unapplied revision in order
# => updates alembic_version after each successful revision
# => Output example:
# => INFO  [alembic.runtime.migration] Context impl PostgreSQLImpl.
# => INFO  [alembic.runtime.migration] Will assume transactional DDL.
# => INFO  [alembic.runtime.migration] Running upgrade  -> abc123de, create users table
# => INFO  [alembic.runtime.migration] Running upgrade abc123de -> def456gh, add email index
 
# Apply up to a specific revision (partial upgrade)
alembic upgrade def456gh
# => applies only up to revision "def456gh"
# => stops there even if newer revisions exist
# => useful for staged deployments or testing specific revisions
 
# Apply a relative number of revisions forward
alembic upgrade +2
# => applies the next 2 pending revisions from current position
# => useful for incremental testing
 
# Preview SQL without executing (offline / dry-run mode)
alembic upgrade head --sql
# => prints all SQL statements to stdout without connecting to the database
# => used for DBA review before execution in regulated environments
# => output can be redirected: alembic upgrade head --sql > migration.sql

# Roll back exactly one revision (most recent)
alembic downgrade -1
# => finds the current revision in alembic_version
# => calls downgrade() for that one revision
# => updates alembic_version to the previous revision
# => Output example:
# => INFO  [alembic.runtime.migration] Running downgrade def456gh -> abc123de, add email index
 
# Roll back two revisions at once
alembic downgrade -2
# => calls downgrade() for the two most recent revisions, in reverse order
# => useful when two related revisions need to be rolled back together
 
# Downgrade to a specific revision
alembic downgrade abc123de
# => rolls back all revisions AFTER "abc123de" until current is "abc123de"
# => calls each downgrade() in reverse order
 
# Downgrade all the way to before the first migration
alembic downgrade base
# => "base" is a special alias meaning "no revisions applied"
# => calls downgrade() for every applied revision in reverse order
# => alembic_version table row is deleted (or set to NULL)
# => WARNING: this destroys all tables created by migrations
 
# Preview downgrade SQL without executing
alembic downgrade -1 --sql
# => prints the downgrade SQL to stdout; does not execute

# Show the current revision of the default database
alembic current
# => connects to the database configured in alembic.ini / env.py
# => reads alembic_version table
# => Output when at a specific revision:
# => INFO  [alembic.runtime.migration] Context impl PostgreSQLImpl.
# => INFO  [alembic.runtime.migration] Will assume transactional DDL.
# => def456gh (head)
# => "(head)" suffix: this revision IS the latest; database is fully up to date
 
# Output when behind the latest revision:
# => abc123de
# => (no "(head)" suffix): revisions exist that have not been applied yet
 
# Output when no migrations have been applied:
# => (no output after INFO lines)
# => or: INFO  [alembic.runtime.migration] No current version.
 
# Verbose output showing full revision details
alembic current --verbose
# => def456gh (head)
# => Rev: def456gh (head)
# => Parent: abc123de
# => Path: alembic/versions/def456gh_add_email_index.py
# => 2026-03-27 10:15:00.000000
# => add email index

# Show all revisions (oldest first by default)
alembic history
# => abc123de -> def456gh (head), add email index
# => <base> -> abc123de, create users table
# => "(head)" marks the latest revision
# => "<base>" marks the starting point (no parent)
# => arrows show direction: parent -> child
 
# Show history in reverse order (newest first)
alembic history --rev-range head:base
# => def456gh -> <head>, add email index
# => abc123de -> def456gh, create users table
# => reading from top to bottom shows newest to oldest
 
# Show a specific range of revisions
alembic history -r abc123de:def456gh
# => shows only revisions between abc123de and def456gh (inclusive)
 
# Verbose history with full details
alembic history --verbose
# => Rev: def456gh (head)
# => Parent: abc123de
# => Path: alembic/versions/def456gh_add_email_index.py
# => 2026-03-27 10:15:00.000000
# => add email index
# =>
# => Rev: abc123de
# => Parent: <base>
# => Path: alembic/versions/abc123de_create_users_table.py
# => 2026-03-27 10:00:00.000000
# => create users table
 
# Show history indicating which revisions have been applied
alembic history --indicate-current
# => def456gh (head), add email index  (current)
# => abc123de, create users table
# => "(current)" marks the revision present in alembic_version

# Create a revision with a descriptive message
alembic revision -m "add status column to orders"
# => -m "add status column to orders": the migration description
# => generates: alembic/versions/abc789xy_add_status_column_to_orders.py
# => slug is derived from the message: spaces -> underscores, lowercase
 
# Good message patterns (imperative mood, specific):
alembic revision -m "create products table"
alembic revision -m "add price index to products"
alembic revision -m "alter users email to varchar 255"
alembic revision -m "drop legacy_code column from products"
alembic revision -m "add foreign key orders to users"
# => each message clearly states the DDL operation being performed
 
# Bad message patterns (vague, non-descriptive):
alembic revision -m "update"
alembic revision -m "fix"
alembic revision -m "migration 3"
# => these show up in alembic history as meaningless entries
# => make debugging and auditing much harder
 
# Viewing the message in history
alembic history
# => abc789xy (head), add status column to orders
# => abc123de, create users table
# => messages appear directly in history output

# alembic/versions/abc789xy_add_status_column_to_orders.py
 
"""add status column to orders
 
Revision ID: abc789xy
Revises: abc123de
Create Date: 2026-03-27 11:00:00.000000
 
"""
 
from alembic import op
import sqlalchemy as sa
 
revision: str = "abc789xy"
down_revision: str | None = "abc123de"
# => down_revision is set automatically to the current head at the time of creation
# => this links this revision to its parent; NEVER change this manually
 
branch_labels = None
depends_on = None
 
 
def upgrade() -> None:
    op.add_column(
        "orders",
        sa.Column("status", sa.String(20), nullable=False, server_default="pending"),
        # => status VARCHAR(20) NOT NULL DEFAULT 'pending'
    )
 
 
def downgrade() -> None:
    op.drop_column("orders", "status")
    # => reverses the add_column

# src/myapp/infrastructure/models.py — SQLAlchemy ORM models
 
from sqlalchemy import Column, Integer, String, DateTime, func
from sqlalchemy.orm import DeclarativeBase
 
 
class Base(DeclarativeBase):
    # => DeclarativeBase: SQLAlchemy 2.0 base class for all ORM models
    # => Base.metadata: MetaData object that tracks all mapped Table objects
    pass
 
 
class User(Base):
    __tablename__ = "users"
    # => tells SQLAlchemy the corresponding table name
 
    id = Column(Integer, primary_key=True, autoincrement=True)
    # => id INTEGER PRIMARY KEY AUTOINCREMENT (SERIAL in PostgreSQL)
 
    username = Column(String(50), nullable=False, unique=True)
    # => username VARCHAR(50) NOT NULL UNIQUE
 
    email = Column(String(255), nullable=False, unique=True)
    # => email VARCHAR(255) NOT NULL UNIQUE
 
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    # => created_at TIMESTAMPTZ DEFAULT NOW()
    # => func.now() is the SQLAlchemy-level representation of the NOW() SQL function

# alembic/env.py — relevant section connecting metadata
 
import sys
from pathlib import Path
 
# Add src/ to sys.path so the import below works without installing the package
_src_path = Path(__file__).parents[2] / "src"
if str(_src_path) not in sys.path:
    sys.path.insert(0, str(_src_path))
 
from myapp.infrastructure.models import Base  # noqa: E402
# => import the Base class that holds all table metadata
 
target_metadata = Base.metadata
# => assign to target_metadata so Alembic knows what to compare against
# => all models that inherit from Base are automatically included
# => if you forget this, --autogenerate produces empty revision files
 
# Autogenerate will now detect:
# => tables that exist in models but not in the database (need CREATE TABLE)
# => tables that exist in the database but not in models (need DROP TABLE — be careful)
# => column additions, removals, and type changes
# => index additions and removals
# => constraint additions and removals (some limitations apply)

# Generate a revision by comparing models to the database
alembic revision --autogenerate -m "add products table"
# => connects to the database (must be running and accessible)
# => reads target_metadata from env.py (must be set correctly)
# => compares database schema to SQLAlchemy models
# => generates a revision file with detected differences
# => Output:
# => INFO  [alembic.autogenerate.compare] Detected added table 'products'
# => INFO  [alembic.autogenerate.compare] Detected added index 'ix_products_sku' on 'products.sku'
# => Generating alembic/versions/xyz789ab_add_products_table.py ...  done

# alembic/versions/xyz789ab_add_products_table.py
 
"""add products table
 
Revision ID: xyz789ab
Revises: def456gh
Create Date: 2026-03-27 12:00:00.000000
 
"""
 
from alembic import op
import sqlalchemy as sa
 
revision: str = "xyz789ab"
down_revision: str | None = "def456gh"
# => automatically set to the current head; links this revision to the chain
branch_labels = None
depends_on = None
 
 
def upgrade() -> None:
    # => AUTOGENERATED: do not edit manually unless you understand the consequences
    op.create_table(
        "products",
        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
        sa.Column("name", sa.String(length=200), nullable=False),
        sa.Column("price", sa.Numeric(precision=10, scale=2), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        # => autogenerate translates primary_key=True to an explicit PrimaryKeyConstraint
    )
    op.create_index(op.f("ix_products_sku"), "products", ["sku"], unique=False)
    # => op.f("ix_products_sku"): uses Alembic's naming convention from MetaData if configured
 
 
def downgrade() -> None:
    # => AUTOGENERATED: drop in reverse order of creation
    op.drop_index(op.f("ix_products_sku"), table_name="products")
    op.drop_table("products")
    # => reverse order: drop index before dropping table

# Inside upgrade() — modifying an existing "users.email" column
 
from alembic import op
import sqlalchemy as sa
 
def upgrade() -> None:
    # Change nullability: make email NOT NULL
    op.alter_column(
        "users",           # => table name
        "email",           # => column name
        nullable=False,    # => change from nullable to NOT NULL
        # => emits: ALTER TABLE users ALTER COLUMN email SET NOT NULL
        # => WARNING: this fails if any existing rows have NULL in email
        # => backfill NULLs before running this migration
    )
 
    # Change server_default
    op.alter_column(
        "users",
        "status",
        server_default="active",
        # => emits: ALTER TABLE users ALTER COLUMN status SET DEFAULT 'active'
        # => existing rows are unaffected; only new inserts use this default
    )
 
    # Remove server_default
    op.alter_column(
        "users",
        "status",
        server_default=None,
        # => None removes the existing default
        # => emits: ALTER TABLE users ALTER COLUMN status DROP DEFAULT
    )
 
    # Change column type (PostgreSQL-specific; use carefully)
    op.alter_column(
        "products",
        "price",
        type_=sa.Numeric(19, 4),
        # => changes from NUMERIC(10,2) to NUMERIC(19,4)
        # => emits: ALTER TABLE products ALTER COLUMN price TYPE NUMERIC(19,4)
        # => safe if new type is wider; dangerous if narrowing (data truncation)
        postgresql_using="price::numeric(19,4)",
        # => postgresql_using: USING clause for explicit type cast (PostgreSQL-specific)
    )
 
 
def downgrade() -> None:
    op.alter_column("users", "email", nullable=True)
    # => emits: ALTER TABLE users ALTER COLUMN email DROP NOT NULL
    op.alter_column("users", "status", server_default=None)
    # => removes the default we added
    op.alter_column(
        "products",
        "price",
        type_=sa.Numeric(10, 2),
        postgresql_using="price::numeric(10,2)",
    )
    # => reverts type change; may lose precision for values with >2 decimal places

# Inside upgrade() — adding unique constraints to "users"
 
from alembic import op
 
def upgrade() -> None:
    # Single-column unique constraint
    op.create_unique_constraint(
        "uq_users_email",   # => constraint name: convention is "uq_{table}_{column}"
        "users",            # => table name
        ["email"],          # => list of columns; must all be part of the constraint
    )
    # => emits: ALTER TABLE users ADD CONSTRAINT uq_users_email UNIQUE (email)
    # => this fails if duplicate email values already exist in the table
    # => check and clean duplicates before running this migration
 
    # Multi-column unique constraint (composite uniqueness)
    op.create_unique_constraint(
        "uq_users_username_tenant",
        "users",
        ["username", "tenant_id"],
        # => uniqueness is defined on the COMBINATION of (username, tenant_id)
        # => two rows can have the same username if they have different tenant_id values
    )
    # => emits: ALTER TABLE users ADD CONSTRAINT uq_users_username_tenant
    # =>        UNIQUE (username, tenant_id)
 
 
def downgrade() -> None:
    op.drop_constraint("uq_users_username_tenant", "users", type_="unique")
    # => emits: ALTER TABLE users DROP CONSTRAINT uq_users_username_tenant
    op.drop_constraint("uq_users_email", "users", type_="unique")
    # => reverse order: drop most recently added first

# Inside upgrade() — adding check constraints to "products"
 
from alembic import op
import sqlalchemy as sa
 
def upgrade() -> None:
    # Price must be positive
    op.create_check_constraint(
        "ck_products_price_positive",   # => constraint name: "ck_{table}_{description}"
        "products",                      # => table name
        sa.text("price > 0"),            # => CHECK condition as a SQL expression
        # => emits: ALTER TABLE products ADD CONSTRAINT ck_products_price_positive
        # =>        CHECK (price > 0)
        # => rejects INSERT or UPDATE where price <= 0 with a constraint violation
    )
 
    # Status must be one of a fixed set of values (alternative to ENUM type)
    op.create_check_constraint(
        "ck_products_status_valid",
        "products",
        sa.text("status IN ('active', 'inactive', 'discontinued')"),
        # => emits: ALTER TABLE products ADD CONSTRAINT ck_products_status_valid
        # =>        CHECK (status IN ('active', 'inactive', 'discontinued'))
    )
 
    # Range constraint: quantity between 0 and 10000
    op.create_check_constraint(
        "ck_products_quantity_range",
        "products",
        sa.text("quantity >= 0 AND quantity <= 10000"),
        # => emits: CHECK (quantity >= 0 AND quantity <= 10000)
    )
 
 
def downgrade() -> None:
    op.drop_constraint("ck_products_quantity_range", "products", type_="check")
    op.drop_constraint("ck_products_status_valid", "products", type_="check")
    op.drop_constraint("ck_products_price_positive", "products", type_="check")
    # => reverse order: drop in reverse of creation
    # => emits: ALTER TABLE products DROP CONSTRAINT ck_products_price_positive

# Step 1: Add the column with a server_default (single migration, production-safe)
 
from alembic import op
import sqlalchemy as sa
 
def upgrade() -> None:
    # Add NOT NULL column with server_default in one operation
    op.add_column(
        "users",
        sa.Column(
            "account_type",
            sa.String(20),
            nullable=False,
            server_default="standard",
            # => server_default="standard": PostgreSQL sets this value for all existing rows
            # => "standard" is the value written to existing rows at migration time
            # => new rows will also get "standard" unless explicitly set
        ),
    )
    # => emits:
    # => ALTER TABLE users ADD COLUMN account_type VARCHAR(20) NOT NULL DEFAULT 'standard'
    # => all existing rows now have account_type = 'standard'
    # => the column is NOT NULL; future inserts without this column get "standard"
 
 
def downgrade() -> None:
    op.drop_column("users", "account_type")
    # => removes the column and its default

# Two-step approach: add with default, then remove the default in same migration
 
def upgrade() -> None:
    op.add_column(
        "users",
        sa.Column(
            "account_type",
            sa.String(20),
            nullable=False,
            server_default="standard",
            # => add column with default so existing rows have a value
        ),
    )
    # => all existing rows now have account_type = 'standard'
 
    op.alter_column(
        "users",
        "account_type",
        server_default=None,
        # => remove the default; future inserts MUST provide a value
        # => emits: ALTER TABLE users ALTER COLUMN account_type DROP DEFAULT
    )
    # => result: column is NOT NULL with no default
    # => existing rows: account_type = 'standard'
    # => new rows: must provide account_type explicitly
 
 
def downgrade() -> None:
    op.drop_column("users", "account_type")

# Inside upgrade() — creating a table with UUID primary key (PostgreSQL)
 
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
 
def upgrade() -> None:
    op.create_table(
        "sessions",
 
        sa.Column(
            "id",
            postgresql.UUID(as_uuid=True),
            # => postgresql.UUID: uses PostgreSQL's native UUID type (not VARCHAR(36))
            # => as_uuid=True: Python receives uuid.UUID objects, not raw strings
            primary_key=True,
            server_default=sa.text("gen_random_uuid()"),
            # => gen_random_uuid(): PostgreSQL built-in function (available from PostgreSQL 13+)
            # => generates a random UUID v4 for each new row
            # => alternative for older PostgreSQL: uuid_generate_v4() (requires uuid-ossp extension)
            nullable=False,
        ),
 
        sa.Column("user_id", postgresql.UUID(as_uuid=True), nullable=False),
        # => foreign key UUID column (no server_default; must be provided explicitly)
 
        sa.Column("token", sa.String(512), nullable=False),
        # => session token as a string (not UUID; could be JWT or other format)
 
        sa.Column(
            "expires_at",
            sa.DateTime(timezone=True),
            nullable=False,
        ),
        # => expiry timestamp; no default (caller must provide)
    )
    # => emits:
    # => CREATE TABLE sessions (
    # =>   id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
    # =>   user_id UUID NOT NULL,
    # =>   token VARCHAR(512) NOT NULL,
    # =>   expires_at TIMESTAMPTZ NOT NULL
    # => )
 
 
def downgrade() -> None:
    op.drop_table("sessions")
    # => DROP TABLE sessions

# Inside upgrade() — adding standard audit timestamp columns
 
from alembic import op
import sqlalchemy as sa
 
def upgrade() -> None:
    op.create_table(
        "audit_log",
 
        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
        # => id SERIAL PRIMARY KEY
 
        sa.Column("event_type", sa.String(100), nullable=False),
        # => event_type VARCHAR(100) NOT NULL
 
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("NOW()"),
            # => created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
            # => set once at INSERT time; never changes
            # => NOW() returns the current transaction start time (not wall clock)
        ),
 
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("NOW()"),
            # => updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
            # => set to NOW() at INSERT time; application must update on each UPDATE
            # => or use a PostgreSQL trigger to auto-update on each UPDATE
        ),
    )
 
    # Adding audit columns to an existing table (common pattern)
    op.add_column(
        "products",
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("NOW()"),
            # => existing rows get NOW() at migration time
            # => represents "created at migration time" for pre-existing rows
            # => acceptable trade-off; backfilling accurate timestamps requires data analysis
        ),
    )
    op.add_column(
        "products",
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("NOW()"),
        ),
    )
 
 
def downgrade() -> None:
    op.drop_column("products", "updated_at")
    op.drop_column("products", "created_at")
    # => reverse order: drop updated_at before created_at
    op.drop_table("audit_log")

# Inside upgrade() — creating a table with an ENUM column
 
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
 
def upgrade() -> None:
    # Create the ENUM type as a schema object first
    order_status = postgresql.ENUM(
        "pending", "processing", "shipped", "delivered", "cancelled",
        # => the allowed values for this enum type
        name="order_status",
        # => name: the PostgreSQL type name; must be unique within the schema
    )
    order_status.create(op.get_bind())
    # => op.get_bind(): returns the current database connection
    # => emits: CREATE TYPE order_status AS ENUM ('pending', 'processing', 'shipped', 'delivered', 'cancelled')
 
    op.create_table(
        "orders",
        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
        sa.Column(
            "status",
            postgresql.ENUM(
                "pending", "processing", "shipped", "delivered", "cancelled",
                name="order_status",
                create_type=False,
                # => create_type=False: the type already exists; don't try to CREATE it again
            ),
            nullable=False,
            server_default="pending",
            # => status order_status NOT NULL DEFAULT 'pending'
        ),
        sa.Column("total", sa.Numeric(10, 2), nullable=False),
    )
    # => emits: CREATE TABLE orders (
    # =>   id SERIAL PRIMARY KEY,
    # =>   status order_status NOT NULL DEFAULT 'pending',
    # =>   total NUMERIC(10,2) NOT NULL
    # => )
 
 
def downgrade() -> None:
    op.drop_table("orders")
    # => DROP TABLE orders (removes column that uses the type first)
 
    order_status = postgresql.ENUM(name="order_status")
    order_status.drop(op.get_bind())
    # => emits: DROP TYPE order_status
    # => must drop table BEFORE dropping the type (dependency order)

# Inside upgrade() — adding composite indexes for common query patterns
 
from alembic import op
 
def upgrade() -> None:
    # Composite index for queries filtering by user_id AND status
    op.create_index(
        "ix_orders_user_status",  # => index name: "ix_{table}_{col1}_{col2}"
        "orders",                  # => table name
        ["user_id", "status"],     # => column order matters: user_id first (higher cardinality)
        unique=False,
        # => emits: CREATE INDEX ix_orders_user_status ON orders (user_id, status)
        # => helps queries: WHERE user_id = $1 AND status = $2
        # => also helps: WHERE user_id = $1 (can use leftmost column alone)
        # => does NOT help: WHERE status = $2 (rightmost only; needs full scan)
    )
 
    # Composite index for queries filtering by tenant_id AND created_at range
    op.create_index(
        "ix_orders_tenant_created",
        "orders",
        ["tenant_id", "created_at"],
        # => helps: WHERE tenant_id = $1 AND created_at BETWEEN $2 AND $3
        # => also helps ORDER BY: WHERE tenant_id = $1 ORDER BY created_at DESC
    )
 
    # Partial index: index only active orders (PostgreSQL-specific)
    op.create_index(
        "ix_orders_active_user",
        "orders",
        ["user_id"],
        unique=False,
        postgresql_where="status != 'cancelled'",
        # => postgresql_where: partial index condition (PostgreSQL-specific)
        # => emits: CREATE INDEX ix_orders_active_user ON orders (user_id)
        # =>        WHERE status != 'cancelled'
        # => smaller index; only indexes rows where status != 'cancelled'
        # => query planner uses this index when query also has WHERE status != 'cancelled'
    )
 
 
def downgrade() -> None:
    op.drop_index("ix_orders_active_user", table_name="orders")
    op.drop_index("ix_orders_tenant_created", table_name="orders")
    op.drop_index("ix_orders_user_status", table_name="orders")
    # => reverse order: drop in reverse of creation

# Inside upgrade() — creating a user_roles junction table
 
from alembic import op
import sqlalchemy as sa
 
def upgrade() -> None:
    # Create the junction table for users <-> roles (many-to-many)
    op.create_table(
        "user_roles",
 
        sa.Column("user_id", sa.Integer, nullable=False),
        # => user_id INTEGER NOT NULL (FK to users.id)
 
        sa.Column("role_id", sa.Integer, nullable=False),
        # => role_id INTEGER NOT NULL (FK to roles.id)
 
        sa.PrimaryKeyConstraint("user_id", "role_id"),
        # => composite primary key: (user_id, role_id) must be unique
        # => prevents a user from having the same role assigned twice
        # => no surrogate id column needed; the FK pair is the natural key
 
        sa.ForeignKeyConstraint(
            ["user_id"],          # => local columns
            ["users.id"],         # => referenced table.column
            name="fk_user_roles_user_id",
            ondelete="CASCADE",   # => delete user_roles rows when user is deleted
        ),
        # => emits: FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
 
        sa.ForeignKeyConstraint(
            ["role_id"],
            ["roles.id"],
            name="fk_user_roles_role_id",
            ondelete="CASCADE",   # => delete user_roles rows when role is deleted
        ),
        # => emits: FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE
    )
    # => emits: CREATE TABLE user_roles (
    # =>   user_id INTEGER NOT NULL,
    # =>   role_id INTEGER NOT NULL,
    # =>   PRIMARY KEY (user_id, role_id),
    # =>   FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
    # =>   FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE
    # => )
 
    # Index on role_id for reverse lookups: "which users have role X?"
    op.create_index("ix_user_roles_role_id", "user_roles", ["role_id"], unique=False)
    # => composite PK already creates an index on (user_id, role_id)
    # => separate index on role_id supports: SELECT user_id FROM user_roles WHERE role_id = $1
 
 
def downgrade() -> None:
    op.drop_index("ix_user_roles_role_id", table_name="user_roles")
    op.drop_table("user_roles")
    # => DROP TABLE also removes the FK constraints and composite PK

# Inside upgrade() — seeding reference data after creating the roles table
 
from alembic import op
import sqlalchemy as sa
 
def upgrade() -> None:
    roles_table = op.create_table(
        "roles",
        # => op.create_table returns a Table object for immediate use
        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
        sa.Column("name", sa.String(50), nullable=False, unique=True),
        sa.Column("description", sa.String(255), nullable=True),
    )
    # => roles_table is a SQLAlchemy Table object (not None)
    # => can be used directly with op.bulk_insert
 
    op.bulk_insert(
        roles_table,
        # => first argument: the Table object returned by create_table
        # => (or use sa.table("roles", ...) for tables created in previous revisions)
        [
            {"id": 1, "name": "admin", "description": "Full system access"},
            {"id": 2, "name": "manager", "description": "Department management access"},
            {"id": 3, "name": "user", "description": "Standard user access"},
            {"id": 4, "name": "readonly", "description": "Read-only access"},
        ],
        # => list of dicts; each dict is one row
        # => keys must match column names in the Table
        # => emits: INSERT INTO roles (id, name, description) VALUES (...), (...), ...
        # => all rows inserted in a single statement for efficiency
    )
    # => roles table now has 4 rows
 
 
def downgrade() -> None:
    op.drop_table("roles")
    # => DROP TABLE roles (also deletes the seed data)

# Inserting into a table from a previous revision
 
def upgrade() -> None:
    roles_table = sa.table(
        "roles",
        # => sa.table(): lightweight Table reference for use in bulk_insert
        # => does not define full schema; only needs columns being inserted
        sa.column("name", sa.String),
        sa.column("description", sa.String),
    )
    op.bulk_insert(
        roles_table,
        [
            {"name": "superadmin", "description": "Super administrator"},
        ],
        # => id will be auto-generated by the database (autoincrement)
    )

# alembic/versions/abc999zz_add_audit_system.py
 
"""add audit system
 
Revision ID: abc999zz
Revises: xyz789ab
Create Date: 2026-03-27 15:00:00.000000
 
"""
 
from alembic import op
import sqlalchemy as sa
 
revision: str = "abc999zz"
down_revision: str | None = "xyz789ab"
branch_labels = None
depends_on = None
 
 
def upgrade() -> None:
    # Operation 1: Add created_by column to users
    op.add_column(
        "users",
        sa.Column("created_by", sa.String(255), nullable=False, server_default="system"),
        # => created_by VARCHAR(255) NOT NULL DEFAULT 'system'
    )
    # => alembic_version still at "xyz789ab" until entire upgrade() completes
 
    # Operation 2: Add updated_by column to users
    op.add_column(
        "users",
        sa.Column("updated_by", sa.String(255), nullable=False, server_default="system"),
        # => updated_by VARCHAR(255) NOT NULL DEFAULT 'system'
    )
 
    # Operation 3: Create audit_log table
    op.create_table(
        "audit_log",
        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
        sa.Column("table_name", sa.String(100), nullable=False),
        sa.Column("operation", sa.String(10), nullable=False),
        # => operation VARCHAR(10): 'INSERT', 'UPDATE', 'DELETE'
        sa.Column("user_id", sa.Integer, nullable=True),
        sa.Column("changed_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")),
    )
 
    # Operation 4: Add index on audit_log for common queries
    op.create_index("ix_audit_log_table_name", "audit_log", ["table_name"])
    # => all four operations run in a single transaction
    # => if op.create_index fails, all three previous operations are rolled back
    # => alembic_version is updated to "abc999zz" only after ALL operations succeed
 
 
def downgrade() -> None:
    # Reverse operations in EXACT reverse order
    op.drop_index("ix_audit_log_table_name", table_name="audit_log")
    # => 4. drop index first (index depends on table)
    op.drop_table("audit_log")
    # => 3. drop table
    op.drop_column("users", "updated_by")
    # => 2. drop updated_by
    op.drop_column("users", "created_by")
    # => 1. drop created_by
    # => exact reverse of upgrade() order ensures no dependency errors

-- The alembic_version table structure (auto-created by Alembic)
-- emitted SQL when Alembic first creates it:
-- CREATE TABLE alembic_version (
--     version_num VARCHAR(32) NOT NULL,   -- => stores the current revision ID
--     CONSTRAINT alembic_version_pkc PRIMARY KEY (version_num)
-- )
-- => single-row table; only one version_num at a time (for linear history)
-- => multi-head setups can have multiple rows (one per head branch)

# Understanding alembic_version state through Python
 
from sqlalchemy import create_engine, text
 
engine = create_engine("postgresql+psycopg2://user:pass@localhost/mydb")
# => create_engine: standard SQLAlchemy connection factory
 
with engine.connect() as conn:
    # Check current version
    result = conn.execute(text("SELECT version_num FROM alembic_version"))
    # => SELECT version_num FROM alembic_version
    row = result.fetchone()
    if row:
        print(f"Current version: {row[0]}")
        # => Output: Current version: abc999zz
        # => this is the revision ID of the most recently applied migration
    else:
        print("No migrations applied (base state)")
        # => alembic_version table exists but has no rows
        # => happens when alembic stamp base is run or table is empty
 
    # After alembic upgrade head, the table has exactly one row
    # => SELECT version_num FROM alembic_version => ('abc999zz',)
 
    # After alembic downgrade base, the row is deleted
    # => SELECT version_num FROM alembic_version => (empty result set)
 
# Manual version stamping (use when DB was migrated outside Alembic)
# alembic stamp abc999zz
# => does NOT run any migrations
# => just inserts/updates alembic_version to record "abc999zz" as current
# => use when: manually applying SQL, importing a dump that already has the schema
 
# alembic stamp head
# => marks the database as at the latest revision without running any migrations
# => useful for: new databases pre-populated with a schema dump

# Downgrade all migrations to base (empty state)
alembic downgrade base
# => "base" is a special alias meaning "before any revision"
# => calls downgrade() for EVERY applied revision in reverse order
# => example output for a database at revision "abc999zz" with chain abc123de -> abc999zz:
# => INFO  [alembic.runtime.migration] Running downgrade abc999zz -> xyz789ab, add audit system
# => INFO  [alembic.runtime.migration] Running downgrade xyz789ab -> def456gh, add products table
# => INFO  [alembic.runtime.migration] Running downgrade def456gh -> abc123de, add email index
# => INFO  [alembic.runtime.migration] Running downgrade abc123de -> , create users table
# => after completion: alembic_version row is deleted (or set to empty)
 
# Verify the database is at base
alembic current
# => (no output after INFO lines) — no current version
# => or: No current version.
 
# Downgrade to base then upgrade back to head (full reset workflow)
alembic downgrade base && alembic upgrade head
# => destroys and recreates all tables
# => useful in development for testing migration chain completeness
# => NEVER run in production unless intentionally wiping all data
 
# Preview the base downgrade SQL without executing
alembic downgrade base --sql
# => prints all DROP TABLE, DROP INDEX, DROP CONSTRAINT statements to stdout
# => safe to run; does not connect to the database in a destructive way

# Programmatic downgrade to base (for test fixtures or CI teardown)
 
import subprocess
 
def reset_database() -> None:
    """Tear down all migrations and recreate from head (development/test use only)."""
    subprocess.run(
        ["alembic", "downgrade", "base"],
        # => runs: alembic downgrade base
        check=True,
        # => raises CalledProcessError if alembic exits with non-zero code
    )
    # => all tables dropped; alembic_version empty
 
    subprocess.run(
        ["alembic", "upgrade", "head"],
        # => runs: alembic upgrade head (re-applies all migrations)
        check=True,
    )
    # => all tables recreated; alembic_version at latest revision
    # => useful for: resetting test database between test suites