cached-duckdb 0.2.0

Fast in-memory DataFrame cache using DuckDB with SQL query interface

cached_duckdb

Fast in-memory DataFrame cache using DuckDB with SQL query interface.

Overview

cached_duckdb replaces pandas dict-based caching with DuckDB in-memory connections for:

• Columnar storage - 20-30% less RAM than pandas
• SQL queries - Single-pass filter+aggregate operations
• Concurrency - Safe parallel reads during writes
• Zero disk usage - Pure in-memory like pandas

Key Features

• Generic database/table API - Works with any cache-based system
• Two storage modes:
  • single_db: One connection per database (enables cross-table JOINs)
  • per_table_db: One connection per (database, table) pair (fully parallel writes)
• Atomic swap - Readers see 100% old or 100% new data, never partial
• TTL-based expiry - Background cleanup with lazy stale flagging
• Scheduler-managed tables - Bypass TTL for scheduled updates
• Thread-safe operations - Per-database or per-table locking

Installation

From PyPI (Recommended)

pip install cached-duckdb

Install Specific Version

pip install cached-duckdb==0.2.0

With Optional Protected Build Extras

pip install "cached-duckdb[all]"

From Local Source

pip install -r requirements.txt

Or install in development mode:

pip install -e .

Verify installed version:

import cached_duckdb
print(cached_duckdb.__version__)

Quick Start

Basic Usage

from cached_duckdb import DuckDbCacheManager
import pandas as pd

# Initialize cache (singleton)
cache = DuckDbCacheManager()

# Store DataFrame
df = pd.DataFrame({
    'date': ['2026-01-01', '2026-01-02'],
    'amount': [1000, 2000],
    'country': ['USA', 'UK']
})
cache.store(database="client_abc", table="sales_data", df=df)

# Query with SQL filtering
result = cache.query(
    database="client_abc",
    table="sales_data",
    sql_where="amount > 1000 AND country = 'USA'",
    columns=["date", "amount"]
)
print(result)

Advanced Queries

# Get all data
df = cache.query(database="app_x", table="dataset_1")

# Filter with WHERE
df = cache.query(
    database="app_x",
    table="dataset_1",
    sql_where="age > 25 AND country = 'USA'"
)

# Select specific columns
df = cache.query(
    database="app_x",
    table="dataset_1",
    columns=["name", "age", "salary"]
)

# Limit results
df = cache.query(
    database="app_x",
    table="dataset_1",
    sql_where="date >= '2026-01-01'",
    limit=100
)

Cross-Table JOINs (single_db mode)

# Execute raw SQL for complex queries
sql = """
    SELECT s.date, s.amount, o.customer_name
    FROM sales_data s
    JOIN orders o ON s.order_id = o.id
    WHERE s.amount > 1000
"""
result = cache.execute_sql(database="client_abc", sql=sql)

Check if Data Exists

if cache.exists(database="client_abc", table="sales_data"):
    # Data is ready and fresh
    df = cache.query(database="client_abc", table="sales_data")
else:
    # Data missing or stale - reload needed
    df = load_from_source()
    cache.store(database="client_abc", table="sales_data", df=df)

TTL and Scheduler-Managed Tables

# Store with custom TTL
cache.store(
    database="client_abc",
    table="sales_data",
    df=df,
    ttl_minutes=60  # Expires after 60 minutes
)

# Scheduler-managed table (no auto-expiry on reads)
cache.store(
    database="client_abc",
    table="sales_data",
    df=df,
    scheduler_managed=True  # Only scheduler updates it
)

Invalidate Cache

# Invalidate one table
cache.invalidate(database="client_abc", table="sales_data")

# Invalidate all tables for a database
cache.invalidate(database="client_abc")

Get Metadata

# Last updated timestamp
last_updated = cache.get_last_updated(database="client_abc", table="sales_data")
print(f"Last updated: {last_updated}")

# Table info
info = cache.get_table_info(database="client_abc", table="sales_data")
print(f"Rows: {info['row_count']}")
print(f"Columns: {info['columns']}")
print(f"Types: {info['column_types']}")

Documentation

• User Manual
• Environment Variables
• Changelog

Author

sreeyenan ([e-mail])

Version

Current version: 0.2.0

Configuration

Environment Variables

# Storage mode: single_db or per_table_db
CACHED_DUCKDB_DEFAULT_MODE=single_db

# Default TTL in minutes
CACHED_DUCKDB_DEFAULT_TTL_MINUTES=30

# Cleanup thread interval
CACHED_DUCKDB_CLEANUP_INTERVAL_MINUTES=5

# Lock timeout in seconds
CACHED_DUCKDB_LOCK_TIMEOUT_SECONDS=30

# Path to connector config file (optional)
CACHED_DUCKDB_CONFIG_FILE_PATH=/path/to/connector_config.json

# Logger name
CACHED_DUCKDB_LOG_NAME=cached_duckdb

Per-Database Configuration File

Create connector_config.json for per-database settings:

{
  "client_abc": {
    "duck_cache_mode": "per_table_db",
    "default_cache_ttl_minutes": 45,
    "sales_data": {
      "cache_ttl_minutes": 60,
      "scheduler_managed": false
    },
    "live_feed": {
      "cache_ttl_minutes": 0,
      "scheduler_managed": true
    }
  },
  "client_xyz": {
    "duck_cache_mode": "single_db"
  }
}

Priority order:

1. Per-table config in JSON file (highest)
2. Per-database config in JSON file
3. Environment variables
4. Hardcoded defaults (lowest)

Load Configuration

from cached_duckdb import DuckDbCacheConfig, DuckDbCacheManager

# From environment
config = DuckDbCacheConfig.from_env()
cache = DuckDbCacheManager(config)

# From dict
config = DuckDbCacheConfig.from_dict({
    "default_mode": "single_db",
    "default_cache_ttl_minutes": 30,
    "config_file_path": "/path/to/connector_config.json"
})
cache = DuckDbCacheManager(config)

Storage Modes

Mode A: single_db (Default)

• One DuckDB connection per database
• Multiple tables share the same connection
• Enables cross-table SQL JOINs
• Write contention: One lock per database

Use when: Database has few tables (< 20) or need cross-table queries

Mode B: per_table_db

• One DuckDB connection per (database, table) pair
• Each table is fully isolated
• Zero write contention between tables
• Fully parallel writes

Use when: Database has many tables (20+) or high write concurrency

Architecture

DuckDbCacheManager (singleton)
  ├── CacheStore        - Atomic writes, table management
  ├── CacheQuery        - SQL queries, metadata
  ├── TTLRegistry       - TTL tracking, cleanup thread
  └── CacheConfigResolver - Per-database config resolution

Thread Safety

• store(): Write-locked per database or per table
• query(): Lock-free parallel reads
• invalidate(): Write-locked, waits for active readers
• Background cleanup: Minimal locking, uses stale flags

Use Cases

• Multi-tenant web APIs - Cache per tenant with database=tenant_id
• Analytics dashboards - Fast in-memory OLAP queries
• ETL pipelines - Store intermediate DataFrames
• Session managers - Replace pandas dict caching
• Microservices - Shared cache library across services

API Reference

DuckDbCacheManager

• store(database, table, df, ttl_minutes=None, scheduler_managed=False) - Store DataFrame
• query(database, table, sql_where=None, columns=None, limit=None) - Query with filtering
• execute_sql(database, sql) - Execute raw SQL
• exists(database, table) - Check if exists and fresh
• invalidate(database, table=None) - Remove from cache
• get_last_updated(database, table) - Get timestamp
• get_table_info(database, table) - Get metadata
• get_raw_connection(database, table=None) - Get DuckDB connection
• shutdown() - Close all connections

Exceptions

• DuckDbCacheError - Base exception
• DuckDbCacheConfigError - Configuration error
• DuckDbCacheLockError - Lock acquisition failed
• DuckDbCacheNotFoundError - Table not found
• DuckDbCacheStaleError - Data is stale

License

MIT License - see LICENSE file

Author

sreeyenan