dbapi-helper 0.1.0

A helper library for implementing Python DB API 2.0 database drivers

dbapi-helper

A modern Python library that provides the basics for implementing Python DB API 2.0 database drivers. This library handles all the boilerplate, leaving you to implement only the cursor.execute() method.

Perfect for creating database drivers for services with REST/HTTP endpoints that accept SQL queries.

Features

• ✅ Full DB API 2.0 compliance (PEP 249)
• ✅ Complete base implementations of Connection and Cursor
• ✅ All standard exceptions and type objects included
• ✅ Modern Python (3.9+) with full type hints
• ✅ Generic cursor type support for better IDE experience
• ✅ Comprehensive documentation and examples
• ✅ Only requires implementing cursor.execute() method

Installation

pip install dbapi-helper

Quick Start

Here's a minimal example of implementing a database driver:

import requests
from dbapi_helper import Connection, Cursor

class RESTCursor(Cursor):
    def __init__(self, api_url: str):
        super().__init__()
        self.api_url = api_url

    def execute(self, operation: str, parameters=None):
        # Make REST API call
        response = requests.post(
            f"{self.api_url}/query",
            json={"sql": operation, "params": parameters}
        )
        data = response.json()

        # Set description (column metadata)
        self.description = tuple(
            (col["name"], col["type"], None, None, None, None, None)
            for col in data["columns"]
        )

        # Set results iterator
        self._results = iter(data["rows"])

        # Store operation for debugging
        self.operation = operation

        return self

class RESTConnection(Connection[RESTCursor]):
    def __init__(self, api_url: str):
        super().__init__()
        self.api_url = api_url

    def cursor(self) -> RESTCursor:
        cursor = RESTCursor(self.api_url)
        self.cursors.append(cursor)
        return cursor

# Usage
conn = RESTConnection("https://api.example.com")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE age > ?", (25,))

for row in cursor.fetchall():
    print(row)

conn.close()

Detailed Usage

Implementing execute()

The execute() method is the only required method to implement. It must:

1. Execute the operation with the given parameters
2. Set self.description to column metadata (or None for non-SELECT queries)
3. Set self._results to an iterator of result tuples
4. Set self.operation to the operation string
5. Return self

def execute(self, operation: str, parameters=None):
    # 1. Execute the query (your custom logic here)
    result = self.my_query_executor(operation, parameters)

    # 2. Set column metadata
    self.description = tuple(
        (name, type_code, None, None, None, None, None)
        for name, type_code in result.columns
    )

    # 3. Set results iterator
    self._results = iter(result.rows)

    # 4. Store the operation
    self.operation = operation

    # 5. Return self
    return self

Important: When overriding execute(), apply the @check_closed decorator to ensure proper error handling:

from dbapi_helper import Cursor
from dbapi_helper.cursor import check_closed

class MyCursor(Cursor):
    @check_closed  # Important!
    def execute(self, operation: str, parameters=None):
        # Your implementation
        ...
        return self

Implementing cursor()

The cursor() method creates and returns a new cursor instance:

from dbapi_helper import Connection
from dbapi_helper.connection import check_closed

class MyConnection(Connection[MyCursor]):
    @check_closed  # Recommended!
    def cursor(self) -> MyCursor:
        cursor = MyCursor(self)  # Pass connection if needed
        self.cursors.append(cursor)  # Track cursor
        return cursor

Column Description Format

The description attribute is a tuple of 7-item tuples, one per column:

(
    name,          # Column name (str)
    type_code,     # Type code (type object or None)
    display_size,  # Display size (int or None)
    internal_size, # Internal size (int or None)
    precision,     # Precision (int or None)
    scale,         # Scale (int or None)
    null_ok,       # Nullable (bool or None)
)

Most implementations only set name and type_code, leaving others as None.

Using Context Managers

Connections support context managers for automatic cleanup:

with RESTConnection("https://api.example.com") as conn:
    cursor = conn.execute("SELECT * FROM users")
    print(cursor.fetchall())
# Connection is automatically committed and closed

Exception Handling

All DB API 2.0 exceptions are included:

from dbapi_helper import (
    Error,              # Base exception
    InterfaceError,     # Interface-related errors
    DatabaseError,      # Database-related errors (base for below)
    DataError,          # Data processing errors
    OperationalError,   # Database operation errors
    IntegrityError,     # Relational integrity errors
    InternalError,      # Database internal errors
    ProgrammingError,   # Programming errors
    NotSupportedError,  # Feature not supported
)

try:
    cursor.execute("INVALID SQL")
except ProgrammingError as e:
    print(f"SQL error: {e}")

Type Objects

Use type objects to describe column types:

from dbapi_helper import STRING, BINARY, NUMBER, DATETIME, ROWID

# In your execute() implementation:
self.description = (
    ("id", NUMBER, None, None, None, None, None),
    ("name", STRING, None, None, None, None, None),
    ("created_at", DATETIME, None, None, None, None, None),
)

Complete Example: REST API Driver

from typing import Any
import requests
from dbapi_helper import Connection, Cursor, OperationalError

class RestAPICursor(Cursor):
    """Cursor for a REST API database."""

    def __init__(self, connection: "RestAPIConnection"):
        super().__init__()
        self.connection = connection

    def execute(self, operation: str, parameters: tuple[Any, ...] | None = None):
        """Execute SQL via REST API."""
        try:
            response = requests.post(
                f"{self.connection.base_url}/query",
                json={
                    "sql": operation,
                    "params": list(parameters) if parameters else []
                },
                headers={"Authorization": f"Bearer {self.connection.token}"},
                timeout=30,
            )
            response.raise_for_status()
            data = response.json()
        except requests.RequestException as e:
            raise OperationalError(f"Query failed: {e}")

        # Set column descriptions
        if data.get("columns"):
            self.description = tuple(
                (
                    col["name"],
                    self._map_type(col.get("type")),
                    None,
                    None,
                    None,
                    None,
                    None,
                )
                for col in data["columns"]
            )
        else:
            self.description = None

        # Set results
        self._results = iter(tuple(row) for row in data.get("rows", []))
        self.operation = operation

        return self

    @staticmethod
    def _map_type(type_name: str | None) -> type | None:
        """Map API type names to Python types."""
        if not type_name:
            return None
        type_map = {
            "string": str,
            "integer": int,
            "float": float,
            "boolean": bool,
            "timestamp": str,  # or datetime
        }
        return type_map.get(type_name.lower())

class RestAPIConnection(Connection[RestAPICursor]):
    """Connection to a REST API database."""

    def __init__(self, base_url: str, token: str):
        super().__init__()
        self.base_url = base_url.rstrip("/")
        self.token = token

    def cursor(self) -> RestAPICursor:
        """Create a new cursor."""
        cursor = RestAPICursor(self)
        self.cursors.append(cursor)
        return cursor

# Usage
def main():
    conn = RestAPIConnection(
        base_url="https://api.example.com",
        token="your-api-token"
    )

    try:
        # Using cursor directly
        cursor = conn.cursor()
        cursor.execute("SELECT id, name, email FROM users WHERE active = ?", (True,))

        print(f"Columns: {[desc[0] for desc in cursor.description]}")

        for row in cursor:
            print(row)

        # Using connection.execute() shortcut
        cursor2 = conn.execute("SELECT COUNT(*) FROM users")
        count = cursor2.fetchone()[0]
        print(f"Total users: {count}")

    finally:
        conn.close()

if __name__ == "__main__":
    main()

API Reference

Connection

• cursor() → Cursor: Create a new cursor (must implement)
• close(): Close connection and all cursors
• commit(): Commit transaction (no-op by default)
• rollback(): Rollback transaction (no-op by default)
• execute(operation, parameters): Convenience method

Cursor

• execute(operation, parameters) → Cursor: Execute query (must implement)
• fetchone() → tuple | None: Fetch next row
• fetchmany(size) → list[tuple]: Fetch multiple rows
• fetchall() → list[tuple]: Fetch all remaining rows
• close(): Close cursor
• executemany(): Not supported (raises NotSupportedError)
• setinputsizes(): No-op (compatibility)
• setoutputsizes(): No-op (compatibility)

Attributes

• cursor.description: Column metadata tuple
• cursor.rowcount: Number of rows affected/returned
• cursor.arraysize: Rows to fetch with fetchmany() (default: 1)
• connection.cursors: List of created cursors

Module Attributes

import dbapi_helper

print(dbapi_helper.apilevel)      # "2.0"
print(dbapi_helper.threadsafety)  # 2
print(dbapi_helper.paramstyle)    # "qmark"

Examples

The examples/ directory contains complete, working implementations:

REST API Driver (examples/rest_driver.py)

A simple REST API database driver that demonstrates the basic pattern for implementing a DB API 2.0 driver. This example shows how to:

• Make HTTP requests to query a REST API
• Map API responses to DB API 2.0 structures
• Handle column descriptions and type mapping

Run with: PYTHONPATH=src python examples/rest_driver.py

Apache Druid Driver (examples/druid_driver.py)

A production-ready implementation for Apache Druid's SQL API. This example demonstrates:

• Streaming large result sets from HTTP responses
• Incremental JSON parsing for memory efficiency
• Multiple authentication methods (Basic Auth, JWT)
• Parameter escaping with pyformat style (%(name)s)
• Type inference from result data
• Context managers and connection pooling

Run with: PYTHONPATH=src python examples/druid_driver.py

Both examples serve as templates you can adapt for your own database drivers.

Development

# Clone the repository
git clone https://github.com/yourusername/dbapi-helper.git
cd dbapi-helper

# Install in development mode
pip install -e .

# Run tests
pytest

# Type checking
mypy src/dbapi_helper

License

Apache License 2.0 - see LICENSE file for details.

Credits

Based on the generic DB API 2.0 implementation from Shillelagh by Beto Dealmeida.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.