Alembic Integration¶

This page covers Alembic migration support and query tracing utilities.

Overview¶

sqlalchemy-cubrid provides Alembic integration through the CubridImpl class, which is automatically registered as an entry point. This enables Alembic to generate and execute CUBRID-compatible DDL during migrations.

Setup¶

Entry point registration¶

The dialect registers itself via pyproject.toml:

[tool.poetry.plugins."alembic.ddl"]
cubrid = "sqlalchemy_cubrid.alembic_impl:CubridImpl"

When Alembic detects a CUBRID connection URL, it automatically uses CubridImpl.

Alembic configuration¶

A standard alembic.ini works with CUBRID:

[alembic]
script_location = alembic
sqlalchemy.url = cubrid://dba:@localhost:33000/testdb

Initialize Alembic¶

alembic init alembic

Then edit alembic/env.py to import your models:

from myapp.models import Base
target_metadata = Base.metadata

CubridImpl¶

The CubridImpl class extends Alembic's DefaultImpl with CUBRID-specific behavior.

transactional_ddl = False¶

Warning

CUBRID auto-commits DDL statements. This means failed migrations cannot be rolled back. Each DDL statement (CREATE TABLE, ALTER TABLE, DROP TABLE, etc.) is committed immediately.

Because of this, CubridImpl sets transactional_ddl = False to inform Alembic that DDL operations are non-transactional.

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

render_type()¶

The render_type() method handles CUBRID collection types in autogenerated migration scripts. When Alembic generates a migration that includes a CubridSet, CubridMultiset, or CubridList column, the method produces importable Python code:

# Autogenerated migration will contain:
from sqlalchemy_cubrid.types import CubridSet

op.add_column("products", sa.Column("tags", CubridSet("VARCHAR(50)")))

For standard SQLAlchemy types, render_type() delegates to the parent class.

compare_type()¶

The compare_type() method provides intelligent type comparison for collection types during alembic --autogenerate:

Same collection kind, same element type -- no change detected
Same collection kind, different element type -- change detected (case-insensitive comparison)
Different collection kinds (e.g., SET vs LIST) -- change detected
Collection vs non-collection -- change detected
Standard types -- delegates to the parent class

# Example: These are considered equal (case-insensitive):
# Database:  CubridSet("varchar(50)")
# Model:     CubridSet("VARCHAR(50)")

# Example: These trigger a migration:
# Database:  CubridSet("VARCHAR(50)")
# Model:     CubridSet("VARCHAR(100)")

Migration Workflow¶

Create a migration¶

alembic revision --autogenerate -m "add products table"

Review the generated migration¶

"""add products table

Revision ID: a1b2c3d4e5f6
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy_cubrid.types import CubridSet


def upgrade():
    op.create_table(
        "products",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("name", sa.String(100), nullable=False),
        sa.Column("tags", CubridSet("VARCHAR(50)")),
    )


def downgrade():
    op.drop_table("products")

Apply the migration¶

alembic upgrade head

Rollback¶

alembic downgrade -1

Warning

Since CUBRID auto-commits DDL, a failed upgrade may leave the database in a partially migrated state. Always review migrations carefully before applying.

Best Practices for CUBRID Migrations¶

Keep migrations small. Since DDL is auto-committed, each migration should be a single logical change. If a multi-step migration fails partway through, manual cleanup will be needed.
Test migrations on a copy. Always test against a non-production database first.
Avoid ALTER TABLE on large tables. CUBRID's ALTER TABLE can be slow on large tables. Consider creating a new table and migrating data instead.
Check for AUTO_INCREMENT limits. Remember that CUBRID allows only one AUTO_INCREMENT column per table.
Use explicit serial names. When using sequences (serials), give them explicit names so Alembic can track them correctly.

Query Tracing¶

sqlalchemy-cubrid includes built-in query tracing utilities that use CUBRID's SET TRACE ON / SHOW TRACE commands. These are useful for debugging performance issues and understanding query execution plans.

trace_query()¶

A one-shot function that enables tracing, executes a query, and returns both the result and the trace output:

from sqlalchemy_cubrid import trace_query
from sqlalchemy import text

with engine.connect() as conn:
    result, trace_output = trace_query(
        conn,
        "SELECT * FROM employees WHERE department_id = :dept",
        {"dept": 10},
    )
    print(trace_output)
    # Trace Statistics:
    #   SELECT (time: 1, fetch: 0, ioread: 0)
    #   ...

Parameters¶

Parameter	Type	Default	Description
`connection`	Connection	required	A SQLAlchemy connection
`sql`	str/text	required	SQL string or text() construct
`params`	dict	`None`	Bind parameters
`output`	str	`"TEXT"`	Output format: `"TEXT"` or `"JSON"`

JSON output¶

result, trace_json = trace_query(
    conn,
    "SELECT * FROM employees",
    output="JSON",
)

import json
trace_data = json.loads(trace_json)
print(json.dumps(trace_data, indent=2))

QueryTracer context manager¶

For tracing multiple queries in a block:

from sqlalchemy_cubrid import QueryTracer
from sqlalchemy import text

with engine.connect() as conn:
    with QueryTracer(conn, output="TEXT") as tracer:
        conn.execute(text("SELECT * FROM employees"))
        conn.execute(text("SELECT * FROM departments"))

    # Trace output is captured when the context manager exits
    # To get the output, use start/stop explicitly:

    tracer = QueryTracer(conn, output="JSON")
    tracer.start()
    conn.execute(text("SELECT * FROM employees WHERE id = 1"))
    conn.execute(text("UPDATE employees SET name = 'Test' WHERE id = 1"))
    trace_output = tracer.stop()
    print(trace_output)

Manual start/stop¶

tracer = QueryTracer(conn, output="TEXT")
tracer.start()

# Execute multiple statements
conn.execute(text("SELECT ..."))
conn.execute(text("UPDATE ..."))

# Get accumulated trace
trace_output = tracer.stop()
print(trace_output)

How tracing works¶

Internally, the tracing utilities issue these CUBRID commands:

SET TRACE ON OUTPUT TEXT (or JSON) -- enables tracing
Your SQL statements execute normally
SHOW TRACE -- retrieves the accumulated trace data
SET TRACE OFF -- disables tracing

The trace output contains execution plans, timing information, I/O statistics, and other performance metrics from the CUBRID query engine.

Import Reference¶

# Alembic integration (usually automatic via entry point)
from sqlalchemy_cubrid.alembic_impl import CubridImpl

# Query tracing
from sqlalchemy_cubrid import trace_query, QueryTracer

# Or from the submodule
from sqlalchemy_cubrid.trace import trace_query, QueryTracer