pycubrid

Pure Python DB-API 2.0 driver for the CUBRID database — no C extensions, no compilation, implements the PEP 249 (DB-API 2.0) interface.

🇰🇷 한국어 · 🇺🇸 English · 🇨🇳 中文 · 🇮🇳 हिन्दी · 🇩🇪 Deutsch · 🇷🇺 Русский

Status: Stable (1.x). The public API follows semantic versioning: minor releases add backward-compatible features and patch releases ship bug fixes; breaking changes are reserved for the next major version (2.0+) and gated by an automated compat-check CI job against api-baseline.json. The one exception is the opt-in pycubrid.compat namespaces: names first released in 1.9.0 or later are provisional and may change in a minor release, announced in the changelog. Active development continues — see RELEASE_POLICY.md for the full contract.

Why pycubrid?

CUBRID is a high-performance open-source relational database, widely adopted in Korean public-sector and enterprise applications. The existing C-extension driver (CUBRIDdb) had build dependencies and platform compatibility issues.

pycubrid solves these problems:

• Pure Python implementation — no C build dependencies, install with pip install only
• Implements PEP 249 (DB-API 2.0) — standard exception hierarchy, type objects, cursor interface
• Offline regression suite with a 95% coverage floor on full coverage runs — see measured coverage and the CI execution policy; routine PR smoke does not measure coverage
• TLS/SSL for sync and async connections — opt-in ssl=True (verified context, TLS 1.2 minimum) or custom ssl.SSLContext on connect() and pycubrid.aio.connect(). On Python 3.10 the async path automatically runs a preflight TLS verification probe to work around a known CPython asyncio bug (fixed in 3.13/3.14) where loop.start_tls() would otherwise hang on cert-verify failures — see Troubleshooting and #156.
• Native asyncio support — async/await API via pycubrid.aio for high-concurrency applications
• PEP 561 typed package — py.typed marker for modern IDE and static analysis support
• Direct CUBRID CAS protocol implementation — no additional middleware required
• LOB (CLOB/BLOB) support — handle large text and binary data

Requirements

Python 3.10 support retirement: Python 3.10 reached upstream end of life on 2026-10-01 (PEP 619). The current 1.8.x line and the upcoming 1.9.x advance-notice release retain Python 3.10 support. The following minor release (planned 1.10.0) will require Python 3.11 or newer, after the 1.9.0 notice has shipped. Upgrade your interpreter, recreate your virtual environment and validate your application before upgrading to that release. This notice does not change the current installation requirement or add a runtime warning.

• Python 3.10+
• CUBRID database server 10.2+ (CI validates 10.2, 11.0, 11.2, 11.4)

Installation

pip install pycubrid

Quick Start

Basic Connection

import pycubrid

conn = pycubrid.connect(
    host="localhost",
    port=33000,
    database="testdb",
    user="dba",
    password="",
)

cur = conn.cursor()
cur.execute("SELECT 1 + 1")
print(cur.fetchone())  # (2,)

cur.close()
conn.close()

Context Manager

import pycubrid

with pycubrid.connect(host="localhost", port=33000, database="testdb", user="dba") as conn:
    with conn.cursor() as cur:
        cur.execute("CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100))")
        cur.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
        conn.commit()

        cur.execute("SELECT * FROM users")
        for row in cur:
            print(row)

Async

import asyncio
import pycubrid.aio

async def main():
    conn = await pycubrid.aio.connect(
        host="localhost", port=33000, database="testdb", user="dba"
    )
    cur = conn.cursor()
    await cur.execute("SELECT 1 + 1")
    print(await cur.fetchone())  # (2,)
    await cur.close()
    await conn.close()

asyncio.run(main())

Practical Examples

Run the setup, the three examples and cleanup in order in the same Python session against a scratch database. Adjust the connection settings for your server. The dedicated table must not already exist: setup deliberately fails rather than reusing or deleting an existing table. Manual commit mode and the initial DDL commit keep rollback focused on the DML below.

Setup

import pycubrid

conn = pycubrid.connect(host="localhost", port=33000, database="testdb", user="dba")
conn.autocommit = False
cur = conn.cursor()
cur.execute("CREATE TABLE pycubrid_quickstart_327 (id INT PRIMARY KEY, name VARCHAR(100))")
conn.commit()

Basic CRUD

cur.execute("INSERT INTO pycubrid_quickstart_327 (id, name) VALUES (?, ?)", (1, "Alice"))
cur.execute("SELECT name FROM pycubrid_quickstart_327 WHERE id = ?", (1,))
print(cur.fetchone())  # ('Alice',)
cur.execute("UPDATE pycubrid_quickstart_327 SET name = ? WHERE id = ?", ("Alicia", 1))
cur.execute("SELECT name FROM pycubrid_quickstart_327 WHERE id = ?", (1,))
print(cur.fetchone())  # ('Alicia',)
cur.execute("DELETE FROM pycubrid_quickstart_327 WHERE id = ?", (1,))
conn.commit()

Transactions: commit and rollback

cur.execute("INSERT INTO pycubrid_quickstart_327 (id, name) VALUES (?, ?)", (2, "Bob"))
conn.commit()
cur.execute("UPDATE pycubrid_quickstart_327 SET name = ? WHERE id = ?", ("Temporary", 2))
conn.rollback()
cur.execute("SELECT name FROM pycubrid_quickstart_327 WHERE id = ?", (2,))
print(cur.fetchone())  # ('Bob',)
conn.rollback()

Error handling

from pycubrid.exceptions import ProgrammingError

try:
    cur.execute("SELECT missing_column FROM pycubrid_quickstart_327 WHERE id = ?", (2,))
except ProgrammingError as exc:
    conn.rollback()
    print(f"Query failed: {exc}")

Cleanup

After setup succeeds, run this cleanup even if an example fails; it removes only the table created above. For more examples, see EXAMPLES.md.

conn.rollback()
cur.execute("DROP TABLE pycubrid_quickstart_327")
conn.commit()
cur.close()
conn.close()

PEP 249 Compliance

• Full standard exception hierarchy: Warning, Error, InterfaceError, DatabaseError, DataError, OperationalError, IntegrityError, InternalError, ProgrammingError, NotSupportedError — also exposed as attributes on Connection/AsyncConnection (PEP 249 optional extension), e.g. conn.IntegrityError is pycubrid.IntegrityError
• Standard type objects: STRING, BINARY, NUMBER, DATETIME, ROWID
• Standard constructors: Date(), Time(), Timestamp(), Binary(), DateFromTicks(), TimeFromTicks(), TimestampFromTicks()
• nextset() raises NotSupportedError (CUBRID does not support multiple result sets)

Features

• Pure Python — no C extensions, no compilation, works everywhere Python runs
• Complete DB-API 2.0 — connect(), Cursor, fetchone/many/all, executemany, callproc
• Parameterized queries — cursor.execute(sql, params) with driver-side parameter binding (? placeholders escaped and interpolated locally)
• Batch operations — executemany() and executemany_batch() for bulk inserts
• LOB support — create_lob(), read/write CLOB and BLOB columns
• Owned schema introspection — get_schema_info() plus eager fetch_schema_info() or explicit close_schema_info(); the #457 live matrix verifies CLASS/VCLASS/ATTRIBUTE/CONSTRAINT/PRIMARY_KEY/IMPORTED_KEYS/EXPORTED_KEYS on CUBRID 10.2/11.4, not all schema codes or native-driver parity
• Auto-commit control — connection.autocommit property for transaction management
• Server version detection — connection.get_server_version() returns version string (e.g., "11.2.0.0378")
• Iterator protocol — iterate over cursor results with for row in cursor
• Context managers — with statements for both connections and cursors
• Async support — pycubrid.aio.connect() with AsyncConnection and AsyncCursor for asyncio event loops
• Per-cursor fetch size — cursor.fetch_size property to tune server-side fetch batch size per cursor

Supported CUBRID Versions

The project targets CUBRID 10.2+ (protocol-compatible). CI validates against:

• 10.2
• 11.0
• 11.2
• 11.4

SQLAlchemy Integration

pycubrid works as a driver for sqlalchemy-cubrid — the SQLAlchemy 2.0 dialect for CUBRID:

pip install "sqlalchemy-cubrid[pycubrid]"

from sqlalchemy import create_engine, text

engine = create_engine("cubrid+pycubrid://dba@localhost:33000/testdb")

with engine.connect() as conn:
    result = conn.execute(text("SELECT 1"))
    print(result.scalar())

SQLAlchemy features (ORM, Core, Alembic migrations, schema reflection) are accessible through the pycubrid driver when used with sqlalchemy-cubrid.

Documentation

Compatibility

Supported: Python 3.10–3.14 and CUBRID 10.2, 11.0, 11.2, 11.4. Ordinary PRs use one Ubuntu/Python 3.12 offline smoke lane; high-risk changes add Python 3.14/CUBRID 11.4. Main/changed-weekly validation uses oldest/newest live endpoints. The full 5 × 4 live matrix runs on explicit dispatch and every release. See CI execution policy and support matrix.

Architecture

graph TD
    app[Application]
    pycubrid[pycubrid Connection/Cursor]
    cas[CAS Protocol]
    server[CUBRID Server]

    app --> pycubrid
    pycubrid --> cas
    cas --> server

graph TD
    root[pycubrid/]
    init["__init__.py - Public API connect(), types, exceptions, __version__"]
    connection[connection.py - Connection class connect/commit/rollback/cursor/LOB]
    cursor[cursor.py - Cursor class execute/fetch/executemany/callproc/iterator]
    types[types.py - DB-API 2.0 type objects and constructors]
    exceptions[exceptions.py - PEP 249 exception hierarchy]
    constants[constants.py - CAS function codes, data types, protocol constants]
    protocol["protocol.py - CAS wire protocol packet classes (20 packet types)"]
    packet[packet.py - Low-level packet reader/writer]
    lob[lob.py - LOB support]
    typed[py.typed - PEP 561 marker]

    root --> init
    root --> connection
    root --> cursor
    root --> types
    root --> exceptions
    root --> constants
    root --> protocol
    root --> packet
    root --> lob
    root --> typed
    root --> aio
    aio["aio/ - AsyncConnection, AsyncCursor, async connect()"]

FAQ

How do I connect to CUBRID with Python?

import pycubrid
conn = pycubrid.connect(host="localhost", port=33000, database="testdb", user="dba")

How do I install pycubrid?

pip install pycubrid — no C extensions or build tools required.

What parameter style does pycubrid use?

Question mark (qmark) style: cursor.execute("SELECT * FROM users WHERE id = ?", (1,))

Parameters are bound driver-side — pycubrid escapes and interpolates values into the SQL string locally before sending the final query to the server. This is not server-side prepared statement binding. The escaping logic is type-aware (strings, bytes, dates, decimals, None → NULL) and safe against SQL injection when used correctly via ? placeholders. Never use f-strings or % formatting to inject untrusted parameter values into SQL — always pass them through ? placeholders. Using f-strings to assemble the SQL structure itself from trusted inputs (for example, expanding an IN-clause with ','.join('?' * len(ids))) is fine. See docs/PARAMETER_BINDING.md for the full 1.x contract — per-type mapping, escaping modes, and explicit non-guarantees.

Does pycubrid work with SQLAlchemy?

Yes. Install pip install "sqlalchemy-cubrid[pycubrid]" and use the connection URL cubrid+pycubrid://dba@localhost:33000/testdb.

What Python versions are supported?

Python 3.10, 3.11, 3.12, 3.13, and 3.14.

Does pycubrid support LOBs (CLOB/BLOB)?

Yes. Insert strings/bytes directly into CLOB/BLOB columns. For reading, LOB columns return data that can be accessed through the cursor.

Is pycubrid thread-safe?

pycubrid has threadsafety = 1, meaning connections cannot be shared between threads. Create a separate connection per thread.

What CUBRID versions are supported?

CUBRID 10.2, 11.0, 11.2, and 11.4 are tested in CI.

Does pycubrid support async/await?

Yes. Use pycubrid.aio.connect() for native asyncio support. The async surface is similar to the sync API: await conn.ping(reconnect=...) performs the same native CHECK_CAS health check as sync Connection.ping(), create_lob() remains sync-only, and auto-commit changes use await conn.set_autocommit(...) instead of a property setter.

Related Projects

• sqlalchemy-cubrid — SQLAlchemy 2.0 dialect for CUBRID
• cubrid-cookbook-python — Production-ready Python examples for CUBRID

Roadmap

See ROADMAP.md for this project's direction and next milestones.

For the ecosystem-wide view, see the CUBRID Labs Ecosystem Roadmap.

Contributing

See CONTRIBUTING.md for guidelines and docs/DEVELOPMENT.md for development setup.

First contribution

New to CUBRID? Pick the repository that matches what you want to work on:

• Documentation and runnable examples: cubrid-cookbook-python
• Pure-Python driver fixes: pycubrid
• SQLAlchemy dialect fixes: sqlalchemy-cubrid

Most first issues can be developed and tested with the offline checks in CONTRIBUTING.md — no Docker or CUBRID server needed. Live CUBRID verification can be completed by CI and maintainers.

Browse open good first issue tasks.

Security

Report vulnerabilities via email — see SECURITY.md. Do not open public issues for security concerns.

Acknowledgments

During pycubrid's initial development, the official CUBRID Node.js driver — node-cubrid (© 2008–2012 Search Solution Corporation, BSD-3-Clause) — was consulted as a reference implementation to understand CUBRID's CAS (Common Application Server) wire protocol: its packet structure and function codes. pycubrid is an independent pure-Python implementation; see NOTICE for details.

We also thank the maintainers and contributors of CUBRID/cubrid-python. Its tests3 suite informed the SQL-feature and type/data regression scenarios added in #409, #410 and #433: INDEX, PARTITION, VIEW, TRIGGER, collections, ENUM and LOB round-trips. These scenarios were adapted to pycubrid's APIs and documented differences, not a claim of full CUBRIDdb API parity. See NOTICE for the source files and third-party notes for provenance and licensing limits.

Disclaimer

This project is part of CUBRID Lab, an independent open-source initiative for CUBRID developer tooling, and is not affiliated with, sponsored by, or endorsed by CUBRID Corporation or the official CUBRID project.

License

MIT — see LICENSE.