tessl/pypi-chdb

chDB is an in-process SQL OLAP Engine powered by ClickHouse

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

DB API 2.0 Interface

Name: tessl/pypi-chdb
Author: tessl

Python Database API 2.0 compliant interface providing standard database connectivity patterns for chDB. This interface enables integration with existing database workflows, connection pooling, and tools that expect standard Python database behavior.

Capabilities

Connection Management

Create and manage database connections with standard DB API methods.

def connect(*args, **kwargs):
    """
    Create a new database connection.
    
    Parameters:
    - cursorclass: Custom cursor class (default: Cursor)
    - path: Optional database path for persistent storage
    
    Returns:
    Connection: Database connection object
    """

def Connect(*args, **kwargs):
    """Alias for connect() function."""

Connection = connect  # Alias for compatibility

Connection Class

Standard database connection with transaction support and cursor creation.

class Connection:
    def __init__(self, cursorclass=None, path: str = None):
        """
        Initialize database connection.
        
        Parameters:
        - cursorclass: Custom cursor class to use
        - path: Optional folder path to store database files
        """
    
    def close(self):
        """
        Close the database connection.
        
        Raises:
        Error: If connection is already closed
        """
    
    def commit(self):
        """Commit current transaction (no-op in chDB)."""
    
    def rollback(self):
        """Roll back current transaction (no-op in chDB)."""
    
    def cursor(self, cursor=None):
        """
        Create a new cursor for executing queries.
        
        Parameters:
        - cursor: Optional cursor class to use
        
        Returns:
        Cursor: Database cursor object
        """
    
    @property
    def open(self) -> bool:
        """Return True if connection is open."""

Cursor Operations

Execute queries and fetch results using standard cursor interface.

class Cursor:
    # Properties
    max_stmt_length: int = 1024000  # Maximum statement size for executemany
    arraysize: int = 1              # Default number of rows for fetchmany
    
    def __init__(self, connection):
        """Initialize cursor with database connection."""
    
    def execute(self, query: str, args=None):
        """
        Execute a database query.
        
        Parameters:
        - query: SQL query string with optional placeholders (%s or %(name)s)
        - args: Query parameters (tuple, list, or dict)
        
        Returns:
        int: Number of affected rows
        """
    
    def executemany(self, query: str, seq_of_parameters):
        """
        Execute query multiple times with different parameters.
        Optimized for bulk INSERT and REPLACE operations.
        
        Parameters:
        - query: SQL query string
        - seq_of_parameters: Sequence of parameter tuples/dicts
        
        Returns:
        int: Total number of affected rows
        """
    
    def callproc(self, procname: str, args=tuple()):
        """
        Execute stored procedure (returns original args).
        
        Parameters:
        - procname: Name of procedure to execute
        - args: Sequence of parameters for procedure
        
        Returns:
        tuple: Original arguments (procedure outputs via server variables)
        """
    
    def fetchone(self):
        """
        Fetch next row of query result set.
        
        Returns:
        tuple: Next row or None if no more rows
        """
    
    def fetchmany(self, size: int = None):
        """
        Fetch multiple rows from query result set.
        
        Parameters:
        - size: Number of rows to fetch (uses arraysize if None)
        
        Returns:
        list: List of row tuples
        """
    
    def fetchall(self):
        """
        Fetch all remaining rows from query result set.
        
        Returns:
        list: List of all row tuples
        """
    
    def nextset(self):
        """
        Move to next result set (if multiple result sets available).
        
        Returns:
        bool: True if more result sets available, False otherwise
        """
    
    def mogrify(self, query: str, args=None) -> str:
        """
        Return exact string that would be sent to database.
        
        Parameters:
        - query: SQL query string with placeholders
        - args: Parameters to substitute
        
        Returns:
        str: Formatted query string
        """
    
    def close(self):
        """Close the cursor and exhaust remaining data."""
    
    # Properties
    @property
    def description(self):
        """Column description tuple for last executed query."""
    
    @property
    def rowcount(self) -> int:
        """Number of rows affected by last operation."""
    
    @property
    def lastrowid(self):
        """ID of last inserted row."""
    
    @property
    def rownumber(self) -> int:
        """Current row number in result set."""

Type Objects and Constants

DB API 2.0 compliant type objects for field type checking.

# API Level Constants
apilevel: str = "2.0"
threadsafety: int = 1
paramstyle: str = "format"

# Type Objects
STRING: DBAPISet     # String field types
BINARY: DBAPISet     # Binary field types  
NUMBER: DBAPISet     # Numeric field types
DATE: DBAPISet       # Date field types
TIME: DBAPISet       # Time field types
TIMESTAMP: DBAPISet  # Timestamp field types
DATETIME: DBAPISet   # Datetime field types (alias for TIMESTAMP)
ROWID: DBAPISet      # Row ID field types

# Special Values
NULL: str = "NULL"

Utility Functions

Helper functions for type conversion and client information.

def Binary(x):
    """
    Convert value to binary type.
    
    Parameters:
    - x: Value to convert
    
    Returns:
    bytes: Binary representation
    """

def get_client_info() -> str:
    """
    Get client version information.
    
    Returns:
    str: Client version string
    """

Exception Classes

DB API 2.0 compliant exception hierarchy for error handling.

class StandardError(Exception):
    """Base exception for chDB operations."""

class Warning(StandardError):
    """Exception for important warnings like data truncations."""

class Error(StandardError):
    """Base class for all error exceptions (not Warning)."""

class InterfaceError(Error):
    """Exception for database interface related errors."""

class DatabaseError(Error):
    """Exception for database related errors."""

class DataError(DatabaseError):
    """Exception for data processing errors (division by zero, etc.)."""

class OperationalError(DatabaseError):
    """Exception for database operational errors (disconnect, etc.)."""

class IntegrityError(DatabaseError):
    """Exception for relational integrity violations (foreign key, etc.)."""

class InternalError(DatabaseError):
    """Exception for internal database errors (invalid cursor, etc.)."""

class ProgrammingError(DatabaseError):
    """Exception for programming errors (table not found, syntax error, etc.)."""

class NotSupportedError(DatabaseError):
    """Exception for unsupported operations or database features."""

Custom Classes

class DBAPISet(frozenset):
    """Custom set class for DB API type comparisons."""
    
    def __ne__(self, other): ...
    def __eq__(self, other): ...
    def __hash__(self): ...

Usage Examples

Basic Connection and Query

import chdb.dbapi as dbapi

# Create connection
conn = dbapi.connect()

# Create cursor and execute query
cur = conn.cursor()
cur.execute('SELECT version()')

# Get column information
print("Columns:", cur.description)

# Fetch results
result = cur.fetchone()
print("Result:", result)

# Clean up
cur.close()
conn.close()

Context Manager Usage

import chdb.dbapi as dbapi

# Using connection as context manager
with dbapi.connect() as conn:
    cur = conn.cursor()
    cur.execute('SELECT 1 as id, "test" as name')
    
    # Fetch all results
    rows = cur.fetchall()
    for row in rows:
        print(row)

Persistent Database Path

import chdb.dbapi as dbapi

# Connect with persistent storage
conn = dbapi.connect(path="/tmp/my_chdb")

cur = conn.cursor()

# Create table
cur.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id Int32,
        name String,
        email String
    ) ENGINE = Memory
''')

# Insert data
cur.execute("INSERT INTO users VALUES (1, 'Alice', 'alice@example.com')")
cur.execute("INSERT INTO users VALUES (2, 'Bob', 'bob@example.com')")

# Query data
cur.execute("SELECT * FROM users ORDER BY id")
users = cur.fetchall()

for user in users:
    print(f"ID: {user[0]}, Name: {user[1]}, Email: {user[2]}")

cur.close()
conn.close()

Working with Different Data Types

import chdb.dbapi as dbapi

conn = dbapi.connect()
cur = conn.cursor()

# Query with various data types
cur.execute('''
    SELECT 
        42 as int_val,
        3.14 as float_val,
        'hello' as string_val,
        today() as date_val,
        now() as timestamp_val
''')

# Check column descriptions
for col in cur.description:
    print(f"Column: {col[0]}, Type: {col[1]}")

# Fetch and display data
row = cur.fetchone()
print("Values:", row)

cur.close()
conn.close()

Using fetchmany()

import chdb.dbapi as dbapi

conn = dbapi.connect()
cur = conn.cursor()

# Execute query that returns multiple rows
cur.execute('''
    SELECT number, number * 2 as doubled 
    FROM numbers(100)
''')

# Fetch rows in batches
batch_size = 10
while True:
    rows = cur.fetchmany(batch_size)
    if not rows:
        break
    
    print(f"Batch of {len(rows)} rows:")
    for row in rows:
        print(f"  {row[0]} -> {row[1]}")

cur.close()
conn.close()

Client Information

import chdb.dbapi as dbapi

# Get client version information
client_info = dbapi.get_client_info()
print(f"chDB client version: {client_info}")

# Check API level and threading support
print(f"API Level: {dbapi.apilevel}")
print(f"Thread Safety: {dbapi.threadsafety}")
print(f"Parameter Style: {dbapi.paramstyle}")

Install with Tessl CLI