Skip to content

pycubrid

🌐 Community translation of README.md — English is canonical; this translation may lag behind the original. The translation-sync CI check flags drift, and maintainers open resync PRs.

Чистый Python-драйвер DB-API 2.0 для базы данных CUBRID — без C-расширений, без компиляции, реализует интерфейс PEP 249 (DB-API 2.0).

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

PyPI version python version ci workflow integration-full workflow coverage license GitHub stars docs


Статус: Stable (1.x). Публичный API следует semantic versioning: минорные релизы добавляют обратно совместимые функции, патч-релизы содержат только исправления ошибок; ломающие изменения отложены до следующего мажорного релиза (2.0+) и контролируются автоматическим CI-джобом compat-check против api-baseline.json. Активная разработка продолжается — полный контракт см. в RELEASE_POLICY.md.

Почему pycubrid?

CUBRID — это высокопроизводительная реляционная база данных с открытым исходным кодом, широко используемая в корейском государственном секторе и корпоративных приложениях. Существующий драйвер на основе C-расширения (CUBRIDdb) имел зависимости сборки и проблемы совместимости с платформами.

pycubrid решает эти проблемы:

  • Чистая реализация на Python — без зависимостей от C-сборки, установка только через pip install
  • Реализует PEP 249 (DB-API 2.0) — стандартная иерархия исключений, объекты типов и интерфейс курсора
  • 770 офлайн-тестов / 811 всего при 97,29 % покрытия кода — большинство тестов запускаются без базы данных
  • TLS/SSL для синхронных и асинхронных подключений — опционально ssl=True (проверенный контекст, минимум TLS 1.2) или пользовательский ssl.SSLContext в connect() и pycubrid.aio.connect(). Примечание: В Python 3.10 async TLS может зависнуть при ошибках проверки сертификата (a known CPython asyncio TLS handshake bug on Python 3.10, исправлено в 3.13/3.14). См. Troubleshooting и #156.
  • Нативная поддержка asyncio — async/await API через pycubrid.aio для приложений с высокой конкурентностью
  • Типизированный пакет PEP 561 — маркер py.typed для современных IDE и инструментов статического анализа
  • Прямая реализация протокола CUBRID CAS — без дополнительного промежуточного ПО
  • Поддержка LOB (CLOB/BLOB) — работа с большими текстовыми и бинарными данными

Требования

  • Python 3.10+
  • Сервер базы данных CUBRID 10.2+

Установка

pip install pycubrid

Быстрый старт

Базовое подключение

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()

Контекстный менеджер

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())

Привязка параметров

# Стиль qmark (знак вопроса)
cur.execute("SELECT * FROM users WHERE name = ? AND age > ?", ("Alice", 25))

# Пакетная вставка с executemany
data = [("Alice", 30), ("Bob", 25), ("Charlie", 35)]
cur.executemany("INSERT INTO users (name, age) VALUES (?, ?)", data)
conn.commit()

Параметризованные запросы

sql = "SELECT * FROM users WHERE department = ?"

cur.execute(sql, ("Engineering",))
engineers = cur.fetchall()

cur.execute(sql, ("Marketing",))
marketers = cur.fetchall()

Соответствие PEP 249

Атрибут Значение
apilevel "2.0"
threadsafety 1 (соединения нельзя разделять между потоками)
paramstyle "qmark" (позиционные параметры ?)
  • Полная стандартная иерархия исключений: Warning, Error, InterfaceError, DatabaseError, OperationalError, IntegrityError, InternalError, ProgrammingError, NotSupportedError
  • Стандартные объекты типов: STRING, BINARY, NUMBER, DATETIME, ROWID
  • Стандартные конструкторы: Date(), Time(), Timestamp(), Binary(), DateFromTicks(), TimeFromTicks(), TimestampFromTicks()

Возможности

  • Чистый Python — без C-расширений, без компиляции, работает везде, где запускается Python
  • Полная DB-API 2.0connect(), Cursor, fetchone/many/all, executemany, callproc
  • Параметризованные запросыcursor.execute(sql, params) с серверным PREPARE_AND_EXECUTE
  • Пакетные операцииexecutemany() и executemany_batch() для массовых вставок
  • Поддержка LOBcreate_lob(), чтение и запись столбцов CLOB и BLOB
  • Интроспекция схемыget_schema_info() для таблиц, столбцов, индексов и ограничений
  • Управление автокоммитом — свойство connection.autocommit для управления транзакциями
  • Определение версии сервераconnection.get_server_version() возвращает строку версии (например, "11.2.0.0378")
  • Итератор курсора — можно проходить результаты for row in cursor
  • Контекстные менеджеры — конструкции with для соединений и курсоров
  • Async-поддержкаpycubrid.aio.connect() с AsyncConnection и AsyncCursor для циклов событий asyncio

Поддерживаемые версии CUBRID

Проект ориентирован на CUBRID 10.x и 11.x и валидируется в CI для следующих версий:

  • 10.2
  • 11.0
  • 11.2
  • 11.4

Интеграция с SQLAlchemy

pycubrid работает как драйвер для sqlalchemy-cubrid — диалекта SQLAlchemy 2.0 для 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 (ORM, Core, миграции Alembic, рефлексия схемы) доступны через драйвер pycubrid при использовании с sqlalchemy-cubrid.

Документация

Руководство Описание
Подключение Строки подключения, формат URL, конфигурация
Сопоставление типов Полное сопоставление типов, специфичные для CUBRID типы, типы коллекций
Справочник API Полная документация API — модули, классы, функции
Протокол Справочник по CAS wire protocol
Разработка Среда разработки, тестирование, Docker, покрытие, CI/CD
Примеры Практические примеры использования с кодом
Устранение неполадок Ошибки подключения, проблемы запросов, работа с LOB, отладка

Совместимость

Python 3.10 Python 3.11 Python 3.12 Python 3.13 Python 3.14
Оффлайн-тесты
CUBRID 11.4 -- -- --
CUBRID 11.2 -- -- --
CUBRID 11.0 -- -- --
CUBRID 10.2 -- -- --

CI запускает указанную выше матрицу при каждом PR/push (Python 3.10 + 3.14 как опорные версии × все версии CUBRID). Полная матрица 5 × 4 Python × CUBRID выполняется каждую ночь, при релизах с тегами и вручную через workflow_dispatch.

Архитектура

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 (18 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

Как подключиться к CUBRID из Python?

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

Как установить pycubrid?

pip install pycubrid — без C-расширений и инструментов сборки.

Какой стиль параметров использует pycubrid?

Стиль вопросительного знака (qmark): cursor.execute("SELECT * FROM users WHERE id = ?", (1,))

Работает ли pycubrid с SQLAlchemy?

Да. Установите pip install "sqlalchemy-cubrid[pycubrid]" и используйте URL подключения cubrid+pycubrid://dba@localhost:33000/testdb.

Какие версии Python поддерживаются?

Python 3.10, 3.11, 3.12, 3.13 и 3.14.

Поддерживает ли pycubrid LOB (CLOB/BLOB)?

Да. Можно вставлять строки/байты напрямую в столбцы CLOB/BLOB. При чтении столбцы LOB возвращают данные, доступные через курсор.

Является ли pycubrid потокобезопасным?

У pycubrid threadsafety = 1, то есть соединения нельзя разделять между потоками. Создавайте отдельное соединение для каждого потока.

Какие версии CUBRID поддерживаются?

Версии CUBRID 10.2, 11.0, 11.2 и 11.4 тестируются в CI.

Поддерживает ли pycubrid async/await?

Да. Используйте pycubrid.aio.connect() для нативной поддержки asyncio. Поверхность async API похожа на sync API: await conn.ping(reconnect=...) выполняет тот же нативный health-check CHECK_CAS, что и sync Connection.ping(), create_lob() по-прежнему остаётся только sync-методом, а изменение автокоммита выполняется через await conn.set_autocommit(...), а не через setter свойства.

Связанные проекты

Дорожная карта

См. ROADMAP.md, чтобы узнать о направлении проекта и следующих этапах.

Для обзора по всей экосистеме см. CUBRID Labs Ecosystem Roadmap.

Участие в проекте

См. CONTRIBUTING.md с рекомендациями и docs/DEVELOPMENT.md с настройкой среды разработки.

Безопасность

Сообщайте об уязвимостях по электронной почте — см. SECURITY.md. Не создавайте публичные issues по вопросам безопасности.

Лицензия

MIT — см. LICENSE.