Alembic Migration Support¶

Guide for using Alembic database migrations with the CUBRID dialect.

Table of Contents¶

Installation
Configuration
Running Migrations
CUBRID-Specific Behavior
Limitations & Workarounds
Examples
Troubleshooting

Installation¶

Install sqlalchemy-cubrid with the alembic extra:

pip install sqlalchemy-cubrid[alembic]

This pulls in Alembic ≥ 1.7.2 as a dependency. The CUBRID Alembic implementation (CubridImpl) is registered automatically when the CUBRID dialect loads — no manual configuration is needed.

Note: If you install Alembic separately (pip install alembic), the CUBRID implementation is still registered as long as sqlalchemy-cubrid is installed in the same environment.

Configuration¶

Initialize Alembic¶

alembic init alembic

This creates an alembic/ directory and an alembic.ini configuration file.

Set the Database URL¶

Edit alembic.ini:

[alembic]
sqlalchemy.url = cubrid://dba:password@localhost:33000/demodb

Or set it dynamically in alembic/env.py:

from sqlalchemy import create_engine

def run_migrations_online():
    connectable = create_engine("cubrid://dba:password@localhost:33000/demodb")

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
        )
        with context.begin_transaction():
            context.run_migrations()

env.py Setup¶

The CubridImpl class is registered as soon as any CUBRID URL loads the dialect, so env.py never needs a CUBRID import. Which template to start from depends on the driver:

Synchronous URLs (cubrid://, cubrid+cubriddb://, cubrid+pycubrid://): the standard env.py from alembic init works without modification.
Async URL (cubrid+aiopycubrid://): use Alembic's async template, alembic init -t async <dir>. Its generated env.py also works without modification. The standard template's online path calls the synchronous engine_from_config(), which cannot drive the async driver and fails with sqlalchemy.exc.MissingGreenlet; offline --sql mode works with either template. Verified on CUBRID 11.4 with Alembic 1.7.2 and 1.20.0: upgrade head / downgrade base through an unmodified async-template env.py (covered by test/test_alembic_registration.py).

A minimal env.py for online migrations:

from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool

# Import your models' metadata
from myapp.models import Base

config = context.config
fileConfig(config.config_file_name)
target_metadata = Base.metadata


def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
        )
        with context.begin_transaction():
            context.run_migrations()


run_migrations_online()

Running Migrations¶

Create a Migration¶

# Auto-generate from model changes
alembic revision --autogenerate -m "add users table"

# Create an empty migration
alembic revision -m "custom migration"

Apply Migrations¶

# Upgrade to the latest version
alembic upgrade head

# Upgrade to a specific revision
alembic upgrade abc123

# Downgrade one step
alembic downgrade -1

# Show current revision
alembic current

# Show migration history
alembic history --verbose

CUBRID-Specific Behavior¶

Transactional DDL¶

CUBRID DDL is transactional. With client autocommit off — the dialect turns it off on every connection — CREATE TABLE, ALTER TABLE, DROP TABLE, TRUNCATE, CREATE INDEX (including WITH ONLINE [PARALLEL n]), CREATE VIEW, CREATE SERIAL and RENAME TABLE run inside the current transaction: ROLLBACK undoes them, and they never commit DML issued earlier in the same transaction (see the CUBRID manual, where ROLLBACK WORK undoes an ALTER TABLE … DROP). The only auto-commit behavior in CUBRID is client autocommit (CCI_DEFAULT_AUTOCOMMIT, or csql's default auto-commit mode), which commits after every statement, DDL or DML.

CubridImpl therefore sets transactional_ddl = True, which tells Alembic:

The whole alembic upgrade runs in one transaction by default. If any revision fails, every revision in that run is rolled back, including the alembic_version update, so the database stays at its starting revision.
With transaction_per_migration=True in context.configure(), each revision runs and commits in its own transaction instead: revisions that finished stay applied, and the failing revision is rolled back as a whole.
context.is_transactional_ddl() returns True.

Schema locks. Uncommitted DDL keeps its schema lock on the table until the transaction ends, so other sessions that touch the table wait (as SCH_S_LOCK waits) for the whole transaction; CUBRID's default lock_timeout is unlimited, so they wait indefinitely rather than time out. For long migrations, or migrations on large tables, set transaction_per_migration=True so each revision commits and releases its locks as soon as it finishes:

# env.py
context.configure(
    connection=connection,
    target_metadata=target_metadata,
    transaction_per_migration=True,
)

Alembic's autocommit_block() is not a way to commit part of an upgrade early on CUBRID: it switches the connection to the AUTOCOMMIT isolation level, which the dialect does not accept before #501, and it gives up the upgrade's atomicity. Use transaction_per_migration=True to commit between revisions instead.

Offline (--sql) scripts. CUBRID has no BEGIN statement (csql rejects it with Syntax error: unexpected 'BEGIN'); a transaction starts implicitly. The CUBRID implementation therefore emits no BEGIN; and ends each transaction with COMMIT; (one per upgrade, or one per revision with transaction_per_migration=True). Run the script with both --no-auto-commit and --no-single-line:

csql -u dba demodb --no-auto-commit --no-single-line -i upgrade.sql

--no-auto-commit makes those COMMIT; lines the only commit points; in csql's default auto-commit mode every statement commits on its own. --no-single-line makes csql stop at the first failing statement and exit with status 1, so the open transaction is rolled back: nothing from the whole upgrade (or, with transaction_per_migration=True, from the failing revision) is kept. In csql's default single-line mode, csql reports the error, continues with the next statements, runs the trailing COMMIT; and exits 0, so a failed script can leave partial schema and a bumped alembic_version.

Changed in 1.8.0

Earlier releases set transactional_ddl = False. Alembic still wrapped each revision in its own transaction in online mode, so each revision was already atomic, but a failed upgrade kept the revisions before it. Now the whole upgrade is atomic by default; set transaction_per_migration=True to keep the previous per-revision behavior.

Auto-Registration¶

Alembic picks its migration implementation from a registry keyed by dialect.name; a DefaultImpl subclass adds itself to that registry when its module is imported (CubridImpl.__dialect__ = "cubrid"). Every CUBRID URL — cubrid://, cubrid+cubriddb://, cubrid+pycubrid:// and cubrid+aiopycubrid:// — has dialect.name == "cubrid". sqlalchemy-cubrid imports sqlalchemy_cubrid.alembic_impl before Alembic looks the name up, in one of two ways depending on the installed Alembic:

Alembic	Who imports `alembic_impl`	When
1.18 and later	Alembic, through the `alembic.plugins` entry point that sqlalchemy-cubrid publishes (`_sqlalchemy_cubrid_alembic`)	during `import alembic`
1.7.2 – 1.17.x	`sqlalchemy_cubrid/dialect.py`	when the CUBRID dialect is loaded (engine creation, or the dialect built from the URL in offline `--sql` mode)

Either way no imports or configuration are required in env.py or your migration files. Without Alembic, nothing is registered and nothing is logged.

With Alembic 1.18 and later, loading the CUBRID dialect does not import Alembic, so applications that never run a migration no longer pay its import cost (with Alembic 1.20.0, SQLAlchemy 2.0.54 and Python 3.10, the first create_engine("cubrid+pycubrid://...") in a fresh process takes about 18 ms instead of about 80 ms). The dialect checks the installed package metadata for this, not Alembic itself. If Alembic has already been imported, the metadata cannot be read, or it does not belong to the alembic package that would be imported (for example an older copy earlier on sys.path), it falls back to the direct import, which is harmless.

Importing Alembic 1.18 and later logs INFO lines from the alembic.runtime.plugins logger (setup plugin alembic.autogenerate.schemas, ..., setup plugin alembic.ext.checkconstraint_byname), and with sqlalchemy-cubrid installed also setup plugin sqlalchemy_cubrid. They are Alembic's own messages. Since the dialect no longer imports Alembic on these versions, they appear only in processes that use Alembic, and only when the application routes INFO records to a handler, for example with logging.basicConfig(level=logging.INFO). With Alembic 1.7.2 – 1.17.x the dialect still imports Alembic (about 0.1 s), but those releases log no such lines. The dialect leaves the alembic logger alone, since its level is the application's choice. To hide the lines, raise that logger's level in the application's logging setup:

import logging

logging.getLogger("alembic").setLevel(logging.WARNING)

Alembic 1.18.0 itself logs these lines from a logger literally named __name__ (logging.getLogger("__name__"), fixed in 1.18.1), so the setting above does not hide them on that release. Upgrade to 1.18.1 or later, or also raise logging.getLogger("__name__") to WARNING.

Alembic 1.18 – 1.20 load an alembic.plugins entry point with for mod in entrypoint.load(), while the Alembic documentation describes the entry point value as the plugin module itself; a plain module is not iterable, so the documented form fails every import alembic with TypeError. _sqlalchemy_cubrid_alembic works with both: it is a module that iterates to itself. Alembic does not guard the entry point load, so the plugin is a top-level module that imports only the standard library; loading it never imports the sqlalchemy_cubrid package, whose import can fail (for example after an unsupported SQLAlchemy is installed over it). Its setup() imports sqlalchemy_cubrid.alembic_impl and turns any failure into a RuntimeWarning (or, under -W error, a log record from the sqlalchemy_cubrid.alembic_plugin logger) instead of raising, so it cannot break import alembic for other projects in the same environment. If the registration still fails, Alembic reports KeyError: 'cubrid'; importing sqlalchemy_cubrid.alembic_impl in env.py registers CubridImpl directly.

If Alembic is installed but fails to import (for example Alembic 1.7.0/1.7.1, which raise NameError on SQLAlchemy 2.x), the dialect still loads and emits a single RuntimeWarning saying the Alembic integration is disabled, with the original exception. If a warning filter turns that warning into an error (-W error), the message is logged through the sqlalchemy_cubrid.dialect logger instead, so the dialect still loads. Upgrading to alembic>=1.7.2,<2.0 fixes it.

Versions before this fix declared an alembic.ddl entry point that Alembic never read, so a default env.py failed with KeyError: 'cubrid' unless it imported sqlalchemy_cubrid.alembic_impl explicitly. That import is harmless and can stay.

Implementation Details¶

class CubridImpl(DefaultImpl):
    __dialect__ = "cubrid"
    transactional_ddl = True

    def emit_begin(self):
        # CUBRID has no BEGIN statement; offline scripts only emit COMMIT;
        pass

The implementation inherits all standard Alembic operations from DefaultImpl: - add_column, drop_column - add_constraint, drop_constraint - create_table, drop_table - create_index, drop_index - alter_column (type change, rename, nullable, default, comment — all native; see "alter_column() behavior" below) - bulk_insert

Limitations & Workarounds¶

Creating a UNIQUE index through Alembic requires a live connection so the dialect can inspect CUBRID's FK auto-indexes and detect collisions. Without a connection, including offline alembic upgrade --sql mode (#861), the guard raises CompileError with a live-connection requirement; it does not silently skip validation. Non-unique index generation is unaffected.

✅ ALTER COLUMN TYPE (native)¶

CUBRID supports changing a column's data type in place via MODIFY:

-- This works:
ALTER TABLE users MODIFY COLUMN name BIGINT;

In Alembic, alter_column(type_=...) emits native ALTER TABLE ... MODIFY <col> <definition>. Because CUBRID's MODIFY restates the entire column definition, the dialect reconstructs the full definition (NOT NULL / DEFAULT / AUTO_INCREMENT / COMMENT) from the existing_* metadata Alembic supplies, so existing attributes are not silently dropped:

def upgrade():
    op.alter_column("users", "name", type_=sa.BigInteger())

Lossy conversions may be rejected

Incompatible or truncating conversions may be rejected by the server depending on the alter_table_change_type_strict system parameter (when yes, incompatible/truncating conversions raise an error; when no, CUBRID may silently truncate). For genuinely lossy or unsupported conversions, fall back to batch_alter_table (table recreate):

def upgrade():
    with op.batch_alter_table("users") as batch_op:
        batch_op.alter_column("name", type_=sa.BigInteger())

✅ RENAME COLUMN (native)¶

CUBRID supports renaming columns via RENAME COLUMN:

-- This works:
ALTER TABLE users RENAME COLUMN old_name TO new_name;

In Alembic, alter_column(new_column_name=...) emits native ALTER TABLE ... RENAME COLUMN old TO new. When a rename is combined with a type change, the dialect emits ALTER TABLE ... CHANGE old new <definition> in a single statement:

def upgrade():
    op.alter_column("users", "old_name", new_column_name="new_name")

⚠️ Long migrations hold schema locks¶

DDL is transactional (see Transactional DDL), so a failed upgrade rolls back cleanly. The cost is that each DDL statement keeps its schema lock until the transaction commits. Be aware:

By default the whole upgrade is one transaction, so every table it touches stays locked until the last revision finishes
Use transaction_per_migration=True for long migrations or large tables
Test migrations against a staging database before production
Maintain database backups before running migrations

Collection types in autogenerate¶

SET, MULTISET and SEQUENCE columns render as cubrid_types.SET(...) with from sqlalchemy_cubrid import types as cubrid_types added to the script's imports. A member type defined by the dialect, such as MONETARY, BIT, STRING, TIMESTAMPTZ or DATETIMELTZ, renders with the same cubrid_types. prefix (cubrid_types.MULTISET(cubrid_types.MONETARY())); a generic SQLAlchemy member keeps sa. (#868). Earlier versions wrote sa.MONETARY(), which raised AttributeError when the migration was loaded.

Foreign key referential actions in autogenerate¶

CUBRID's default ON DELETE / ON UPDATE action is RESTRICT, and SHOW CREATE TABLE always prints it, so inspect(...).get_foreign_keys() reports a foreign key created without actions as {"ondelete": "RESTRICT", "onupdate": "RESTRICT"}. Reflection keeps reporting what the server says. For autogenerate, CubridImpl treats that reflected RESTRICT as equal to a model ForeignKey with no ondelete / onupdate (#597), so an unchanged foreign key produces no drop_constraint / create_foreign_key pair, whether the model leaves the action out or names RESTRICT. Adding ondelete="CASCADE" or SET NULL, removing it again, or switching between them is still detected. NO ACTION is different: CUBRID prints it as NO ACTION, but Alembic itself treats NO ACTION and no action as equal in both directions, so switching between them is not detected. CUBRID rejects ON UPDATE CASCADE; use SET NULL, RESTRICT or NO ACTION for onupdate.

`alter_column()` behavior¶

The CUBRID Alembic implementation maps alter_column() to native CUBRID DDL:

type_= changes → MODIFY (or CHANGE when combined with a rename)
new_column_name= renames → RENAME COLUMN (or CHANGE with a type change)
nullable= and comment= changes → MODIFY (or CHANGE with a rename), because CUBRID has no ALTER COLUMN ... SET/DROP NOT NULL and no COMMENT ON COLUMN. They need the column type: pass existing_type= (autogenerate does), or the call raises ValueError naming the column. In a hand-written migration also pass existing_nullable=, existing_server_default= and existing_comment= for every attribute that must survive: on CUBRID 10.2 and 11.0 the MODIFY resets any attribute it is not told about, so a comment-only change without existing_nullable=False drops NOT NULL and the default there
server_default= alone → ALTER COLUMN c SET DEFAULT <value>; server_default=None → ALTER COLUMN c SET DEFAULT NULL, since CUBRID has no DROP DEFAULT (#861)

These definitions are rebuilt from the existing_* metadata so that attributes such as NOT NULL / DEFAULT / COMMENT are preserved. CUBRID 10.2 and 11.0 reset an attribute that MODIFY leaves out, while 11.2 and 11.4 keep it, so the dialect states every attribute it knows and makes changes explicit: NULL when a NOT NULL column becomes nullable (11.2+ rejects NULL on a column that is already nullable with error -709, so it is left out then), and COMMENT '' when comment=None removes a comment. A NOT NULL column cannot have a NULL default on any version, so dropping its default emits two statements: one that makes the column nullable without a default, then one that restores NOT NULL. For a NOT NULL column pass server_default=None with existing_type= and existing_nullable=False (plus existing_comment= to keep a comment); without existing_nullable=False the single SET DEFAULT NULL is rejected by the server, and without existing_type the call raises ValueError.

Important — hand-written type-changing migrations must pass existing_*. Because CUBRID's MODIFY / CHANGE restate the entire column definition, the dialect can only preserve an attribute it is told about. Alembic autogenerate populates existing_type, existing_nullable, existing_server_default, and existing_comment from the reflected column, but a manual op.alter_column(..., type_=...) does not. When editing a migration by hand, pass existing_nullable=, existing_server_default=, existing_comment=, and existing_autoincrement= for any attribute that must survive the type change — otherwise it will be dropped. To intentionally remove an attribute (e.g. a default), pass it explicitly (server_default=None), which overrides the existing_* value.

Summary¶

Operation	Supported	Workaround
`create_table`	✅	—
`drop_table`	✅	—
`add_column`	✅	—
`drop_column`	✅	—
`alter_column` (nullable)	✅	`MODIFY`; needs `existing_type`, plus the other `existing_*` values to keep
`alter_column` (default)	✅	`SET DEFAULT NULL` drops it; for a `NOT NULL` column pass `existing_type` and `existing_nullable=False`
`alter_column` (comment)	✅	`MODIFY`; needs `existing_type`, plus the other `existing_*` values to keep
`alter_column` (type)	✅	`batch_alter_table` for lossy conversions
`alter_column` (rename)	✅	—
`create_index`	✅	Expression, `DESC`, filtered (`cubrid_where`) and prefix (`cubrid_length`) indexes are reflected and compared by autogenerate (#869); `if_not_exists=True` raises `CompileError` (CUBRID has no `CREATE INDEX IF NOT EXISTS`); check `inspect(conn).has_index()` first instead
`drop_index`	✅	Emits `DROP INDEX <name> ON <table>`, so `table_name` is required (`CompileError` without it); `if_exists=True` raises `CompileError` (CUBRID has no `DROP INDEX IF EXISTS`)
`add_constraint`	✅	—
`drop_constraint`	✅	—
`bulk_insert`	✅	—
Transactional DDL	✅	`transaction_per_migration=True` for long or large-table migrations

Examples¶

Create a Table¶

"""create users table

Revision ID: 001
"""
from alembic import op
import sqlalchemy as sa


def upgrade():
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
        sa.Column("name", sa.String(100), nullable=False),
        sa.Column("email", sa.String(200), unique=True),
        sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP")),
    )
    op.create_index("ix_users_email", "users", ["email"])


def downgrade():
    op.drop_index("ix_users_email", table_name="users")
    op.drop_table("users")

Add a Column¶

def upgrade():
    op.add_column("users", sa.Column("is_active", sa.SmallInteger(), server_default="1"))


def downgrade():
    op.drop_column("users", "is_active")

Change Column Type (via batch)¶

def upgrade():
    with op.batch_alter_table("users") as batch_op:
        batch_op.alter_column("name", type_=sa.String(500))


def downgrade():
    with op.batch_alter_table("users") as batch_op:
        batch_op.alter_column("name", type_=sa.String(100))

Rename Column (via batch)¶

def upgrade():
    with op.batch_alter_table("users") as batch_op:
        batch_op.alter_column("name", new_column_name="full_name")


def downgrade():
    with op.batch_alter_table("users") as batch_op:
        batch_op.alter_column("full_name", new_column_name="name")

Troubleshooting¶

"No implementation found for dialect 'cubrid'"¶

Cause: sqlalchemy-cubrid is not installed, or not installed with the alembic extra.

Fix:

pip install sqlalchemy-cubrid[alembic]

`INFO` log lines `setup plugin ...`¶

Alembic 1.18+ logs these while it is imported. Loading the CUBRID dialect no longer imports Alembic on those versions, so they only appear in processes that use Alembic. Set logging.getLogger("alembic").setLevel(logging.WARNING) in the application (on Alembic 1.18.0, which logs them from a logger named __name__, upgrade or quiet that logger too); see Auto-Registration.

"Alembic is required for migration support"¶

Cause: The alembic_impl module was imported directly without Alembic installed.

Fix:

pip install "alembic>=1.7.2,<2.0"

Migration partially applied¶

Cause: CUBRID DDL is transactional, so a failed online upgrade does not leave a half-applied revision: the failing transaction is rolled back. By default that is the whole upgrade; with transaction_per_migration=True it is the failing revision, and the revisions before it stay committed and recorded in alembic_version. Partial state can still come from:

client autocommit, which commits every statement: driver-level autocommit, csql's default auto-commit mode, or isolation_level="AUTOCOMMIT" where the dialect accepts it (#501);
an offline (--sql) script run without csql --no-auto-commit --no-single-line (csql's default single-line mode continues past a failing statement and still runs the trailing COMMIT;);
a revision that calls COMMIT itself (for example through op.execute).

Fix: 1. Manually inspect the database state 2. Either complete the remaining operations manually, or reverse the completed ones 3. Stamp the revision to the correct state: alembic stamp <revision>

`alter_column` type change rejected by the server¶

Cause: A lossy or incompatible type conversion when the alter_table_change_type_strict system parameter is yes.

Fix: For genuinely lossy/unsupported conversions, use batch_alter_table — see ALTER COLUMN TYPE (native).

`alembic revision --autogenerate` fails with a reflection error¶

Cause: Foreign keys are read from SHOW CREATE TABLE. Since #589 a failure there (a dropped connection, an authorization error, a driver error) raises instead of being reported as "this table has no foreign keys", which made autogenerate propose add_fk for foreign keys that already exist.

Fix: Fix the underlying error and run autogenerate again; set pool_pre_ping=True on the engine if the connection went stale. Since CUBRID 11.2 an unqualified name resolves in the current user's schema, so the table and view listings autogenerate compares against contain only the current user's own tables and views (#864); tables of other owners are neither reflected nor proposed for drop_table. Run autogenerate as the owner of the tables it should manage.

Migration Safety Checklist¶

DDL is transactional

CUBRID rolls back DDL with the transaction; only client autocommit commits it early. By default a failed alembic upgrade leaves no partial schema and no version bump. Uncommitted DDL holds schema locks, so use transaction_per_migration=True for long or large-table migrations.

Lossy type changes may be rejected by the server

alter_column(type_=...) and alter_column(new_column_name=...) emit native CUBRID DDL (MODIFY / RENAME COLUMN / CHANGE). Only genuinely lossy or incompatible type conversions are rejected (governed by alter_table_change_type_strict); for those, use op.batch_alter_table() with tested downgrade steps.

Always test upgrade + downgrade on staging

Validate full forward and backward migration chains before production rollout.

Create backups for destructive operations

Back up data before drop_column, drop_table, or multi-step restructuring migrations.

Pre-Migration Checklist¶

Before running migrations in production:

[ ] Plan lock duration — DDL holds schema locks until commit, and by default the whole upgrade is one transaction. Use transaction_per_migration=True for long or large-table migrations.
[ ] No client autocommit — don't run migrations with client autocommit on (driver-level autocommit, csql's default auto-commit mode, or isolation_level="AUTOCOMMIT" where the dialect accepts it (#501)), and run offline scripts with csql --no-auto-commit --no-single-line, so a failure rolls back cleanly.
[ ] Backup database — cubrid backupdb demodb before destructive operations
[ ] Test upgrade + downgrade cycle — run alembic upgrade head && alembic downgrade -1 && alembic upgrade head on staging
[ ] Verify state after each step — query db_class system table to confirm schema matches expectations
[ ] Maintenance window — schedule high-impact migrations during low-traffic periods
[ ] Rollback script ready — for each upgrade(), have a tested manual rollback SQL if downgrade() is insufficient

Recommended Pre-Deploy Sequence¶

cubrid backupdb demodb — create a full backup
alembic upgrade head on a staging copy
Run smoke tests and critical queries against staging
alembic downgrade -1 and alembic upgrade head to verify reversibility
Deploy during a maintenance window for high-impact schema changes
Monitor alembic current post-deploy to confirm expected revision

Advisory CI Safety Check¶

Add the following script to list revisions with several DDL operations. This is advisory (warning-only) and does not block CI. Such revisions still roll back as a whole on failure, but they hold schema locks longer. The script counts calls to Alembic DDL operations, including constraint creation (create_unique_constraint, create_foreign_key, create_check_constraint, create_primary_key), in each upgrade() and downgrade() separately; bare references such as op.drop_table without a call are not counted. It is a heuristic, not a control-flow analysis:

#!/usr/bin/env python3
"""Check Alembic revisions for multiple DDL operations (advisory).

CUBRID DDL is transactional, so a failing revision is rolled back as a
whole. Every DDL statement holds a schema lock on its table until the
transaction commits, though, so this lists revisions with several DDL
calls: they keep tables locked longer and are candidates for running with
``transaction_per_migration=True`` or for splitting.

Only calls such as ``op.create_table(...)`` or ``batch_op.add_column(...)``
count; a bare reference like ``op.drop_table`` is not a DDL operation. This is
a heuristic AST scan, not control-flow analysis: a call in a loop or branch
counts once, as written, and DDL issued from helpers defined outside
``upgrade()``/``downgrade()`` is not seen.

Usage:
    python scripts/alembic_safety_check.py alembic/versions/
"""
from __future__ import annotations

import ast
import sys
from pathlib import Path

DDL_CALLS = {
    "create_table", "drop_table", "add_column", "drop_column",
    "create_index", "drop_index", "alter_column",
    "add_constraint", "drop_constraint",
    "create_unique_constraint", "create_foreign_key",
    "create_check_constraint", "create_primary_key",
}


def check_revision(path: Path) -> list[str]:
    tree = ast.parse(path.read_text(encoding="utf-8"))
    warnings = []
    for func in ast.walk(tree):
        if not isinstance(func, ast.FunctionDef) or func.name not in ("upgrade", "downgrade"):
            continue
        ddl_count = sum(
            1 for node in ast.walk(func)
            if isinstance(node, ast.Call)
            and isinstance(node.func, ast.Attribute)
            and node.func.attr in DDL_CALLS
        )
        if ddl_count > 1:
            warnings.append(
                f"{path.name}:{func.name}() has {ddl_count} DDL operations "
                "(schema locks are held until the transaction commits)"
            )
    return warnings


def main() -> None:
    versions_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("alembic/versions")
    if not versions_dir.is_dir():
        print(f"Directory not found: {versions_dir}")
        sys.exit(1)

    all_warnings = []
    for py_file in sorted(versions_dir.glob("*.py")):
        all_warnings.extend(check_revision(py_file))

    if all_warnings:
        print("⚠️  Alembic safety warnings (advisory):")
        for w in all_warnings:
            print(f"  • {w}")
        print(f"\nTotal: {len(all_warnings)} warning(s)")
        print(
            "Tip: these revisions roll back whole on failure but hold schema locks "
            "until commit; use transaction_per_migration=True for long or "
            "large-table migrations."
        )
    else:
        print("✓ No revision has more than one DDL call per function.")


if __name__ == "__main__":
    main()

Example CI integration (.github/workflows/ci.yml):

- name: Alembic safety check (advisory)
  run: python scripts/alembic_safety_check.py alembic/versions/ || true

Rollback Script Template¶

For critical migrations, create a companion rollback SQL file:

-- rollback_001_add_users_table.sql
-- Manual rollback for revision abc123 (add users table)
-- Use if alembic downgrade fails or is insufficient.

DROP TABLE IF EXISTS users;

-- Verify: SELECT class_name FROM db_class WHERE class_name = 'users';
-- Expected: no rows

Store rollback scripts in alembic/rollbacks/ alongside your versions directory.

See also: Connection Guide · Type Mapping · Feature Support