SQLAlchemy 2.0–2.1 dialect for the CUBRID database — Python ORM, schema reflection, Alembic migrations, and type mapping for SQLAlchemy and CUBRID-specific types.
🇰🇷 한국어 · 🇺🇸 English · 🇨🇳 中文 · 🇮🇳 हिन्दी · 🇩🇪 Deutsch · 🇷🇺 Русский
Status: Beta. The core public API follows semantic versioning; minor releases may add features and bug fixes while the project remains under active development.
CUBRID is a high-performance open-source relational database, widely adopted in Korean public-sector and enterprise applications. Until now, there was no actively maintained SQLAlchemy dialect that supports the modern 2.0–2.1 API.
sqlalchemy-cubrid bridges that gap:
- Full SQLAlchemy 2.0–2.1 dialect with statement caching and PEP 561 typing
- 619 offline tests with ~98.26% code coverage — no database required to run them
- Concurrency stress tests —
QueuePoolsync threaded + asyncio.gather workloads validated against live CUBRID - SQLAlchemy 2.2-ready compat shim — private API access wrapped in
_compat.py(still pinned<2.2until full SA 2.2 validation) - Tested against 4 CUBRID versions (10.2, 11.0, 11.2, 11.4) across Python 3.10 -- 3.14
- CUBRID-specific DML constructs:
ON DUPLICATE KEY UPDATE,MERGE,REPLACE INTO - Alembic migration support out of the box
- Three driver options — C-extension (
cubrid://), pure Python (cubrid+pycubrid://), or async pure Python (cubrid+aiopycubrid://)
flowchart TD
app["Application"] --> sa["SQLAlchemy Core/ORM"]
sa --> dialect["CubridDialect"]
dialect --> pycubrid["pycubrid driver"]
dialect --> cext["CUBRIDdb driver"]
dialect --> aio["pycubrid.aio async driver"]
pycubrid --> server["CUBRID Server"]
cext --> server
aio --> server
flowchart TD
expr["SQL Expression"] --> compiler["CubridSQLCompiler"] --> sql["SQL String"]
- Python 3.10+
- SQLAlchemy 2.0 – 2.1
- CUBRID-Python (C-extension) or pycubrid (pure Python)
pip install sqlalchemy-cubridWith the pure Python driver (no C build needed):
pip install "sqlalchemy-cubrid[pycubrid]"With Alembic support:
pip install "sqlalchemy-cubrid[alembic]"from sqlalchemy import create_engine, text
engine = create_engine("cubrid://dba:password@localhost:33000/demodb")
with engine.connect() as conn:
result = conn.execute(text("SELECT 1"))
print(result.scalar())from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(String(200), unique=True)
engine = create_engine("cubrid://dba:password@localhost:33000/demodb")
Base.metadata.create_all(engine)
with Session(engine) as session:
user = User(name="Alice", email="[email protected]")
session.add(user)
session.commit()from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import text
engine = create_async_engine("cubrid+aiopycubrid://dba:password@localhost:33000/demodb")
async with AsyncSession(engine) as session:
result = await session.execute(text("SELECT 1"))
print(result.scalar())- Type mapping for SQLAlchemy standard and CUBRID-specific types — numeric, string, date/time, bit, LOB, collection, and JSON types
- SQL compilation -- SELECT, JOIN, CAST, LIMIT/OFFSET, subqueries, CTEs, window functions
- DML extensions --
ON DUPLICATE KEY UPDATE,MERGE,REPLACE INTO,FOR UPDATE,TRUNCATE - DDL support --
COMMENT,IF NOT EXISTS/IF EXISTS,AUTO_INCREMENT - Schema reflection -- tables, views, columns, PKs, FKs, indexes, unique constraints, comments
- Alembic migrations via
CubridImpl(auto-discovered entry point) - All 6 CUBRID isolation levels (dual-granularity: class-level + instance-level)
- Async support —
create_async_engine("cubrid+aiopycubrid://...")via pycubrid.aio
- No
RETURNING—INSERT/UPDATE/DELETE ... RETURNINGnot supported; usecursor.lastrowidorLAST_INSERT_ID() - No sequences — CUBRID uses
AUTO_INCREMENTonly - No multi-schema — single schema per database
- DDL auto-commits — migrations are not transactional (
transactional_ddl = False) - SQLAlchemy 2.0–2.1 only — pinned to
<2.2due to internal API dependencies (details) - Async requires pycubrid >= 1.3.2,<2.0 — the
cubrid+aiopycubrid://driver needs the async-capable pycubrid package line currently supported by this project
| Guide | Description |
|---|---|
| Connection | Connection strings, URL format, driver setup, pool tuning |
| Type Mapping | Full type mapping, CUBRID-specific types, collection types |
| DML Extensions | ON DUPLICATE KEY UPDATE, MERGE, REPLACE INTO, query trace |
| Isolation Levels | All 6 CUBRID isolation levels, configuration |
| Alembic Migrations | Setup, configuration, limitations, batch workarounds |
| Feature Support | Comparison with MySQL, PostgreSQL, SQLite |
| ORM Cookbook | Practical ORM examples, relationships, queries |
| Development | Dev setup, testing, Docker, coverage, CI/CD |
| Driver Compatibility | CUBRID-Python driver versions and known issues |
| Troubleshooting | Common issues, error solutions, debugging techniques |
| Async Connection | Async engine setup with cubrid+aiopycubrid:// |
| Component | Supported versions |
|---|---|
| Python | 3.10, 3.11, 3.12, 3.13, 3.14 |
| CUBRID | 10.2, 11.0, 11.2, 11.4 |
| SQLAlchemy | 2.0–2.1 |
| Alembic | >=1.7 |
| pycubrid (sync) | >=1.3.2,<2.0 |
| pycubrid (async) | >=1.3.2,<2.0 |
from sqlalchemy import create_engine
engine = create_engine("cubrid://dba:password@localhost:33000/demodb")For the pure Python driver (no C build needed): create_engine("cubrid+pycubrid://dba@localhost:33000/demodb")
Yes. sqlalchemy-cubrid is built for SQLAlchemy 2.0–2.1 and supports the 2.0-style API including Session.execute(), typed Mapped[] columns, and statement caching.
Yes. Install with pip install "sqlalchemy-cubrid[alembic]". The dialect auto-registers via entry point. Note that CUBRID auto-commits DDL, so migrations are not transactional.
Python 3.10, 3.11, 3.12, 3.13, and 3.14.
No. CUBRID does not support INSERT ... RETURNING or UPDATE ... RETURNING. Use cursor.lastrowid or SELECT LAST_INSERT_ID() instead.
from sqlalchemy_cubrid import insert
stmt = insert(users).values(name="Alice").on_duplicate_key_update(name="Alice Updated")cubrid:// uses the C-extension driver (CUBRIDdb) which requires compilation. cubrid+pycubrid:// uses the pure Python driver which installs with pip alone — no build tools needed. cubrid+aiopycubrid:// uses the async variant of the pure Python driver for use with create_async_engine and AsyncSession.
Yes. Use create_async_engine("cubrid+aiopycubrid://...") with the pycubrid async driver. Requires pycubrid>=1.3.2,<2.0. Both pycubrid dialects use native Connection.ping(False) / AsyncConnection.ping(False) for pool_pre_ping, and all Core and ORM features work with AsyncSession.
- pycubrid — Pure Python DB-API 2.0 driver for CUBRID
- cubrid-cookbook-python — Production-ready Python examples for CUBRID
See ROADMAP.md for this project's direction and next milestones.
For the ecosystem-wide view, see the CUBRID Labs Ecosystem Roadmap and Project Board.
See CONTRIBUTING.md for guidelines and docs/DEVELOPMENT.md for development setup.
Report vulnerabilities via email -- see SECURITY.md. Do not open public issues for security concerns.
MIT -- see LICENSE.