ducklingaling 0.0.3

DB-API 2.0 compliant Python client library for DuckDBLayer

ducklingaling

A Python client for DuckDBLayer - the PostgreSQL-compatible database that stores data as Parquet files.

Key Features:

• 🔌 DB-API 2.0 compliant - Drop-in replacement for psycopg2
• 🦆 DuckLake integration - Time travel, snapshots, and CDC built-in
• ⚡ Async support - Native asyncio with asyncpg backend
• 🐼 DataFrame support - Query directly to pandas
• 🏊 Connection pooling - Thread-safe pool management

---

Table of Contents

• Installation
• Quick Start
• Connection
• Executing Queries
• Transactions
• DuckLake Features
  • Catalog Management
  • Snapshots & Time Travel
  • Change Data Feed (CDC)
  • Metadata Introspection
  • Maintenance Operations
  • S3 Configuration
• Async API
• Connection Pooling
• DataFrame Integration
• Error Handling
• Type Reference
• Migration from psycopg2
• Troubleshooting

---

Installation

# Basic installation (sync only)
pip install ducklingaling

# With async support (adds asyncpg)
pip install ducklingaling[async]

# With pandas DataFrame support
pip install ducklingaling[pandas]

# Full installation (async + pandas)
pip install ducklingaling[all]

# Development installation
pip install ducklingaling[dev]

Requirements: Python 3.9+

---

Quick Start

import ducklingaling

# Connect
conn = ducklingaling.connect(
    host="localhost",
    port=5432,
    user="admin",
    password="secret",
    database="main"
)

# Query
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE active = %s", [True])

for row in cursor:
    print(row)

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

With context managers (recommended):

import ducklingaling

with ducklingaling.connect(host="localhost", user="admin", password="secret") as conn:
    cursor = conn.execute("SELECT COUNT(*) FROM orders")
    count = cursor.fetchone()[0]
    print(f"Total orders: {count}")
# Connection automatically closed

---

Connection

Connection Parameters

conn = ducklingaling.connect(
    host="localhost",      # Server hostname (default: "localhost")
    port=5432,             # Server port (default: 5432)
    database="main",       # Database name (default: "main")
    user="admin",          # Username (optional)
    password="secret",     # Password (optional)
    timeout=30,            # Connection timeout in seconds (optional)
)

Connection Properties

conn.closed       # bool: True if connection is closed
conn.autocommit   # bool: Get/set autocommit mode
conn.isolation_level  # str: Current isolation level

Connection Methods

conn.cursor()           # Create a new cursor
conn.execute(sql, params)  # Execute and return cursor (convenience)
conn.executemany(sql, seq_of_params)  # Batch execute
conn.commit()           # Commit transaction
conn.rollback()         # Rollback transaction
conn.close()            # Close connection

---

Executing Queries

Basic Queries

cursor = conn.cursor()

# Simple query
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()

# Parameterized query (ALWAYS use this for user input!)
cursor.execute(
    "SELECT * FROM users WHERE id = %s AND status = %s",
    [42, "active"]
)
row = cursor.fetchone()

# Named parameters
cursor.execute(
    "SELECT * FROM users WHERE name = %(name)s",
    {"name": "Alice"}
)

Fetching Results

cursor.execute("SELECT * FROM users")

# Fetch one row
row = cursor.fetchone()  # Returns tuple or None

# Fetch multiple rows
rows = cursor.fetchmany(size=10)  # Returns list of tuples

# Fetch all rows
all_rows = cursor.fetchall()  # Returns list of tuples

# Iterate directly
cursor.execute("SELECT * FROM users")
for row in cursor:
    print(row)

Insert, Update, Delete

# Insert
cursor.execute(
    "INSERT INTO users (name, email) VALUES (%s, %s)",
    ["Alice", "[e-mail]"]
)
print(f"Inserted {cursor.rowcount} rows")

# Bulk insert
users = [
    ("Bob", "[e-mail]"),
    ("Charlie", "[e-mail]"),
]
cursor.executemany(
    "INSERT INTO users (name, email) VALUES (%s, %s)",
    users
)

# Update
cursor.execute(
    "UPDATE users SET status = %s WHERE id = %s",
    ["inactive", 42]
)

# Delete
cursor.execute("DELETE FROM users WHERE status = %s", ["deleted"])

# Don't forget to commit!
conn.commit()

Cursor Properties

cursor.description   # Column metadata (name, type_code, ...)
cursor.rowcount      # Number of affected rows
cursor.query         # Last executed query string
cursor.statusmessage # Status message from last operation

---

Transactions

Explicit Transaction Control

conn.autocommit = False  # Default

try:
    cursor = conn.cursor()
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except Exception:
    conn.rollback()
    raise

Transaction Context Manager

with conn.transaction():
    cursor = conn.cursor()
    cursor.execute("INSERT INTO logs (message) VALUES (%s)", ["Started"])
    cursor.execute("UPDATE counters SET value = value + 1")
    # Commits automatically on success
    # Rolls back automatically on exception

Isolation Levels

conn.isolation_level = "READ COMMITTED"      # Default
conn.isolation_level = "REPEATABLE READ"
conn.isolation_level = "SERIALIZABLE"
conn.isolation_level = "READ UNCOMMITTED"

---

DuckLake Features

DuckLake provides versioned data storage with time travel, snapshots, and change tracking. All DuckLake methods are available directly on the connection object.

Catalog Management

# Attach a DuckLake catalog (auto-installs extension)
conn.attach_catalog(
    catalog_path="/data/my_catalog.db",  # Catalog database file
    data_path="/data/parquet/",          # Parquet file directory (optional)
    name="lake"                          # Catalog name (default: "lake")
)

# Switch between catalogs
conn.use_catalog("lake")

# Detach when done
conn.detach_catalog("lake")

Snapshots & Time Travel

Every write creates a new snapshot. Query historical data:

# List all snapshots
snapshots = conn.list_snapshots(catalog="lake")
for snap in snapshots:
    print(f"Snapshot {snap['snapshot_id']} at {snap['timestamp']}")
    print(f"  Tables: {snap['table_count']}")

# Get current snapshot ID
current_id = conn.get_current_snapshot(catalog="lake")

# Query data at a specific version
old_data = conn.query_at_version(
    table="users",
    version=5,
    catalog="lake"
)

# Query data at a specific timestamp
historical = conn.query_at_timestamp(
    table="users",
    timestamp="2025-01-15 10:30:00",
    catalog="lake"
)

Change Data Feed (CDC)

Track exactly what changed between any two points:

# Get changes between snapshot IDs
changes = conn.table_changes(
    table="orders",
    start=10,
    end=20,
    catalog="lake"
)

for change in changes:
    change_type = change['change_type']
    # 'insert'           - New row added
    # 'delete'           - Row removed
    # 'update_preimage'  - Row BEFORE update
    # 'update_postimage' - Row AFTER update

    print(f"{change_type}: order_id={change['id']}")

# Get changes from last week using timestamps
recent = conn.table_changes(
    table="orders",
    start="now() - INTERVAL '7 days'",
    end="now()",
    catalog="lake"
)

# Filter by change type
inserts = [c for c in changes if c['change_type'] == 'insert']
deletes = [c for c in changes if c['change_type'] == 'delete']
updates = [c for c in changes if 'update' in c['change_type']]

Metadata Introspection

Query database structure without raw SQL:

# List all tables
tables = conn.list_tables()
for t in tables:
    print(f"{t['schema_name']}.{t['table_name']}")

# List tables in specific schema
lake_tables = conn.list_tables(schema="lake")

# Describe table columns
columns = conn.describe_table("users")
for col in columns:
    print(f"  {col['column_name']}: {col['data_type']}")
    print(f"    nullable: {col.get('is_nullable')}")

# With schema filter
columns = conn.describe_table("users", schema="lake")

# List all schemas
schemas = conn.list_schemas()  # Returns: ['main', 'lake', ...]

# List all views
views = conn.list_views()
views = conn.list_views(schema="lake")

# List extensions
extensions = conn.list_extensions()
for ext in extensions:
    print(f"{ext['extension_name']}: loaded={ext.get('loaded')}")

Maintenance Operations

Keep your DuckLake catalog healthy:

# Checkpoint - flush writes, persist metadata
conn.checkpoint()

# Expire old snapshots
# Use dry_run=True to preview what would be expired
preview = conn.expire_snapshots(
    catalog="lake",
    older_than_days=30,
    dry_run=True
)
print(f"Would expire {len(preview)} snapshots")

# Actually expire
conn.expire_snapshots(catalog="lake", older_than_days=30)

# Clean up unreferenced files
conn.cleanup_files(catalog="lake", older_than_days=7)

# Delete orphaned files (not in catalog metadata)
conn.delete_orphaned_files(catalog="lake", older_than_days=7)

# Optimize storage
conn.merge_adjacent_files(catalog="lake")   # Combine small files
conn.rewrite_data_files(catalog="lake")     # Rewrite for better compression

S3 Configuration

Configure cloud storage access:

# Set S3 region
conn.configure_s3(region="us-west-2")

# With custom endpoint (MinIO, etc.)
conn.configure_s3(region="us-east-1", endpoint="https://minio.example.com")

# Create S3 credentials

# Option 1: AWS SSO
conn.create_s3_secret(
    credential_source="sso",
    region="us-west-2",
    profile="my-sso-profile"
)

# Option 2: AWS profile (~/.aws/credentials)
conn.create_s3_secret(
    credential_source="profile",
    region="us-west-2",
    profile="default"
)

# Option 3: Explicit credentials
conn.create_s3_secret(
    credential_source="env",
    region="us-west-2",
    key_id="AKIAIOSFODNN7EXAMPLE",
    secret="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
)

# Option 4: IAM role (EC2, ECS, Lambda)
conn.create_s3_secret(
    credential_source="iam",
    region="us-west-2"
)

Extension Management

# Install an extension
conn.install_extension("httpfs")
conn.install_extension("json")

# Load an extension
conn.load_extension("httpfs")

---

Async API

For asyncio applications, use the async interface:

import asyncio
from ducklingaling import aio

async def main():
    # Connect
    conn = await aio.connect(
        host="localhost",
        port=5432,
        user="admin",
        password="secret"
    )

    # Execute and fetch all
    rows = await conn.fetch("SELECT * FROM users WHERE active = $1", True)

    # Fetch single row
    row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", 42)

    # Fetch single value
    count = await conn.fetchval("SELECT COUNT(*) FROM users")

    # Execute without results
    await conn.execute(
        "INSERT INTO users (name) VALUES ($1)",
        "Alice"
    )

    await conn.close()

asyncio.run(main())

Async Context Manager

async with aio.connect(host="localhost", user="admin", password="secret") as conn:
    rows = await conn.fetch("SELECT * FROM users")

Async Transactions

async with aio.connect(...) as conn:
    async with conn.transaction():
        await conn.execute("UPDATE accounts SET balance = balance - $1 WHERE id = $2", 100, 1)
        await conn.execute("UPDATE accounts SET balance = balance + $1 WHERE id = $2", 100, 2)
        # Auto-commits on success, auto-rollbacks on exception

Async Connection Pool

async def main():
    pool = await aio.create_pool(
        host="localhost",
        user="admin",
        password="secret",
        min_size=5,
        max_size=20
    )

    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT * FROM users")

    await pool.close()

---

Connection Pooling

For multi-threaded applications:

import ducklingaling
from concurrent.futures import ThreadPoolExecutor

# Create pool
pool = ducklingaling.create_pool(
    minconn=5,           # Minimum connections
    maxconn=20,          # Maximum connections
    host="localhost",
    user="admin",
    password="secret"
)

def process_order(order_id):
    with pool.connection() as conn:
        cursor = conn.execute(
            "SELECT * FROM orders WHERE id = %s",
            [order_id]
        )
        return cursor.fetchone()

# Use with thread pool
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(process_order, order_ids))

# Close pool when done
pool.closeall()

Pool Properties & Methods

pool.size       # Current number of connections
pool.available  # Number of available connections
pool.in_use     # Number of connections in use
pool.closed     # Whether pool is closed

pool.getconn()  # Get a connection (manual management)
pool.putconn(conn)  # Return a connection
pool.resize(min_size, max_size)  # Resize pool
pool.check()    # Health check
pool.closeall() # Close all connections

---

DataFrame Integration

Requires pandas extra: pip install ducklingaling[pandas]

import ducklingaling

with ducklingaling.connect(...) as conn:
    cursor = conn.execute("SELECT * FROM sales WHERE year = 2024")

    # Fetch as DataFrame
    df = cursor.fetchdf()

    # Now use pandas!
    print(df.head())
    print(df.describe())

    top_products = df.groupby('product')['revenue'].sum().nlargest(10)

Insert from DataFrame

import pandas as pd

df = pd.DataFrame({
    'id': [1, 2, 3],
    'name': ['Alice', 'Bob', 'Charlie'],
    'score': [95.5, 87.3, 92.1]
})

with ducklingaling.connect(...) as conn:
    cursor = conn.cursor()
    cursor.executemany(
        "INSERT INTO students (id, name, score) VALUES (%s, %s, %s)",
        df.values.tolist()
    )
    conn.commit()

---

Error Handling

The library uses DB-API 2.0 standard exceptions:

from ducklingaling import (
    Error,              # Base exception
    Warning,            # Warnings
    InterfaceError,     # Client/interface errors
    DatabaseError,      # Database errors (base)
    DataError,          # Data processing errors
    OperationalError,   # Connection/operational errors
    IntegrityError,     # Constraint violations
    InternalError,      # Internal database errors
    ProgrammingError,   # SQL syntax/programming errors
    NotSupportedError,  # Unsupported operations
)

Exception Hierarchy

Error
├── InterfaceError
└── DatabaseError
    ├── DataError
    ├── OperationalError
    ├── IntegrityError
    ├── InternalError
    ├── ProgrammingError
    └── NotSupportedError

Error Handling Examples

import ducklingaling
from ducklingaling import OperationalError, ProgrammingError, IntegrityError

# Connection errors
try:
    conn = ducklingaling.connect(host="nonexistent", timeout=5)
except OperationalError as e:
    print(f"Could not connect: {e}")

# SQL errors
try:
    cursor.execute("SELEC * FORM users")  # Typos!
except ProgrammingError as e:
    print(f"SQL syntax error: {e}")

# Constraint violations
try:
    cursor.execute("INSERT INTO users (id) VALUES (1)")  # Duplicate key
except IntegrityError as e:
    print(f"Constraint violation: {e}")

# Catch all database errors
try:
    cursor.execute(some_query)
except DatabaseError as e:
    print(f"Database error: {e}")
    conn.rollback()

---

Type Reference

PostgreSQL to Python Type Mapping

PostgreSQL Type	Python Type
INTEGER, BIGINT	int
FLOAT, DOUBLE	float
NUMERIC, DECIMAL	decimal.Decimal
TEXT, VARCHAR	str
BOOLEAN	bool
TIMESTAMP	datetime.datetime
DATE	datetime.date
TIME	datetime.time
JSON, JSONB	dict or list
UUID	uuid.UUID
BYTEA	bytes
ARRAY	list

DB-API Type Objects

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

# Check column types
if cursor.description[0][1] == NUMBER:
    print("First column is numeric")

---

Migration from psycopg2

ducklingaling is designed as a drop-in replacement for psycopg2:

# Before (psycopg2)
import psycopg2
conn = psycopg2.connect(host="localhost", dbname="mydb", user="admin", password="secret")

# After (ducklingaling)
import ducklingaling
conn = ducklingaling.connect(host="localhost", database="mydb", user="admin", password="secret")

Key Differences

Feature	psycopg2	ducklingaling
Parameter name	dbname	database
Backend	PostgreSQL	DuckDBLayer
DuckLake methods	❌	✅ conn.list_snapshots() etc.
DataFrame support	❌	✅ cursor.fetchdf()

What Works the Same

• cursor.execute(), fetchone(), fetchall(), fetchmany()
• conn.commit(), rollback(), close()
• Parameterized queries with %s and %(name)s
• Context managers
• Exception hierarchy
• cursor.description, cursor.rowcount

---

Troubleshooting

Connection Refused

# Check server is running
# Check host/port are correct
# Check firewall allows connection

conn = ducklingaling.connect(
    host="localhost",
    port=5432,
    timeout=10  # Add timeout to fail faster
)

Authentication Failed

# Verify credentials in users.yaml on server
# Check auth method matches (cleartext vs md5)

Query Timeout

# Large queries may timeout
# Configure server-side: query_timeout in config.yaml

Memory Issues with Large Results

# Don't use fetchall() for large results
# Use fetchmany() or iterate:

cursor.execute("SELECT * FROM huge_table")
while True:
    rows = cursor.fetchmany(size=1000)
    if not rows:
        break
    process(rows)

Connection Pool Exhausted

# Increase max pool size
pool = ducklingaling.create_pool(maxconn=50, ...)

# Make sure to return connections
with pool.connection() as conn:
    # Connection returned when block exits
    pass

---

License

MIT License - see LICENSE for details.

Links

• DuckDBLayer Repository
• Full Documentation
• DuckLake
• DB-API 2.0 Specification (PEP 249)