glean-database 0.3.1

Database access classes with caching capabilities

Glean Database

Overview

Glean Database is a unified, async-first Python library providing a consistent interface for interacting with multiple database systems and caching layers. It abstracts away the complexities of different database clients while offering built-in caching capabilities to optimize query performance.

Key Features

• 🔌 Unified Interface: Single API for 7 database systems (SQLite, PostgreSQL, MySQL, MSSQL, Oracle, MongoDB, TimescaleDB)
• ⚡ Async-First: Built on asyncio for high-performance concurrent operations
• 💾 Integrated Caching: Optional caching layer with in-memory and Redis backends
• 🎯 Type-Safe: Full type hints for better IDE support and code safety
• 🧪 Fully Tested: 58% code coverage with comprehensive unit tests
• 📊 Time-Series Support: Native TimescaleDB integration for time-series data
• 🌊 Result Streaming: Memory-efficient streaming for large datasets
• 🏊 Connection Pooling: Built-in connection pooling for all SQL databases

Installation

# Basic installation (includes SQLite, PostgreSQL, MongoDB, TimescaleDB)
pip install glean-database

# With MySQL support
pip install glean-database[mysql]

# With MSSQL support  
pip install glean-database[mssql]

# With Oracle support
pip install glean-database[oracle]

# With all database backends
pip install glean-database[all]

# For development
pip install glean-database[dev]

Quick Start

SQLite

from database import SQLiteDatabase

async def main():
    db = SQLiteDatabase("my_database.db")
    await db.connect()
    
    # Create table
    await db.execute("""
        CREATE TABLE users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            email TEXT NOT NULL
        )
    """)
    
    # Insert data
    await db.execute(
        "INSERT INTO users (id, name, email) VALUES (:id, :name, :email)",
        {"id": 1, "name": "John Doe", "email": "[e-mail]"}
    )
    
    # Query data
    results = await db.query("SELECT * FROM users WHERE id = :id", {"id": 1})
    print(results)
    
    await db.disconnect()

PostgreSQL

from database import PostgresDatabase

async def main():
    db = PostgresDatabase(
        database="mydb",
        host="localhost",
        port=5432,
        username="postgres",
        password="secret"
    )
    await db.connect()
    
    # Use named parameters with :param style
    results = await db.query(
        "SELECT * FROM users WHERE email = :email",
        {"email": "[e-mail]"}
    )
    
    await db.disconnect()

MongoDB

from database import MongoDatabase

async def main():
    db = MongoDatabase(
        database="mydb",
        host="localhost",
        port=27017
    )
    await db.connect()
    
    # Insert documents
    await db.execute(
        "users",
        "insert",
        [{"name": "John Doe", "email": "[e-mail]"}]
    )
    
    # Query documents
    results = await db.query(
        "users",
        {"email": "[e-mail]"},
        {"sort": [("name", 1)]}
    )
    
    await db.disconnect()

MySQL

from database import MySQLDatabase

async def main():
    db = MySQLDatabase(
        database="mydb",
        host="localhost",
        port=3306,
        username="root",
        password="secret"
    )
    await db.connect()
    
    # Use named parameters with :param style
    results = await db.query(
        "SELECT * FROM users WHERE email = :email",
        {"email": "[e-mail]"}
    )
    
    await db.disconnect()

MSSQL (SQL Server)

from database import MSSQLDatabase

async def main():
    db = MSSQLDatabase(
        database="mydb",
        host="localhost",
        port=1433,
        username="sa",
        password="YourStrong@Passw0rd"
    )
    await db.connect()
    
    # Use named parameters
    results = await db.query(
        "SELECT * FROM users WHERE id = :id",
        {"id": 1}
    )
    
    await db.disconnect()

Oracle

from database import OracleDatabase

async def main():
    db = OracleDatabase(
        database="ORCL",
        host="localhost",
        port=1521,
        username="system",
        password="oracle",
        service_name="ORCLPDB1"
    )
    await db.connect()
    
    # Oracle uses :param style natively
    results = await db.query(
        "SELECT * FROM users WHERE user_id = :id",
        {"id": 1}
    )
    
    await db.disconnect()

TimescaleDB

from database import TimescaleDatabase
from datetime import timedelta

async def main():
    db = TimescaleDatabase(
        database="metrics_db",
        host="localhost",
        port=5432,
        username="postgres"
    )
    await db.connect()
    
    # Create hypertable
    await db.execute("""
        CREATE TABLE metrics (
            time TIMESTAMPTZ NOT NULL,
            sensor_id TEXT NOT NULL,
            temperature DOUBLE PRECISION
        )
    """)
    
    await db.create_hypertable(
        "metrics",
        "time",
        chunk_time_interval="1 day"
    )
    
    # Add retention policy (keep data for 30 days)
    await db.add_retention_policy("metrics", "30 days")
    
    # Time bucket aggregation
    results = await db.time_bucket_query(
        bucket_width="1 hour",
        time_column="time",
        table_name="metrics",
        select_columns=["sensor_id"],
        aggregates=["AVG(temperature) as avg_temp"],
        order_by="bucket DESC",
        limit=10
    )
    
    await db.disconnect()

Caching

Add caching to any database backend to improve query performance:

from database import PostgresDatabase, RedisCache, InMemoryCache

# With Redis cache
cache = RedisCache(host="localhost", port=6379)
db = PostgresDatabase(database="mydb", cache=cache)

# Or with in-memory cache
cache = InMemoryCache()
db = PostgresDatabase(database="mydb", cache=cache)

# With background cleanup (recommended for in-memory cache with TTLs)
cache = InMemoryCache(cleanup_interval=60)  # Clean up every 60 seconds
cache.start_cleanup()  # Start background task
db = PostgresDatabase(database="mydb", cache=cache)

# Don't forget to stop cleanup when done
await cache.stop_cleanup()

Caching is automatic - identical queries with the same parameters will be served from cache.

Memory Cache with Automatic Expiration

The InMemoryCache supports automatic cleanup of expired keys:

• On-access expiration: Expired keys are removed when accessed
• Background cleanup: Optional background task periodically removes expired keys
• Configurable interval: Set cleanup interval in seconds (default: 60)
• Max size limit: Limit cache to a maximum number of keys

from database import InMemoryCache
from datetime import timedelta

# Create cache with 30-second cleanup interval
cache = InMemoryCache(cleanup_interval=30)

# Start background cleanup task
cache.start_cleanup()

# Use cache normally
await cache.set("key", "value", ttl=timedelta(minutes=5))

# Stop cleanup when done
await cache.stop_cleanup()

Memory Cache with Size Limits

Limit the cache to a maximum number of keys to prevent unbounded memory growth:

from database import InMemoryCache

# Cache limited to 1000 keys
cache = InMemoryCache(max_size=1000)

# When cache is full, oldest keys are evicted in this priority:
# 1. Expired keys (if any)
# 2. Keys with soonest expiration
# 3. First key inserted (FIFO)

await cache.set("key1", "value1")
await cache.set("key2", "value2", ttl=timedelta(hours=1))
# ... add up to 1000 keys

# Adding 1001st key will automatically evict the oldest
await cache.set("key1001", "value1001")  # Oldest key removed

Query Result Streaming

For large datasets, use streaming to process results without loading everything into memory.

SQLite Streaming

from database import SQLiteDatabase

db = SQLiteDatabase("large_database.db")
await db.connect()

# Stream results one at a time
count = 0
async for row in db.query_stream("SELECT * FROM large_table"):
    process_row(row)  # Process without loading all rows
    count += 1

print(f"Processed {count} rows")

PostgreSQL & TimescaleDB Streaming

from database import PostgresDatabase

db = PostgresDatabase(database="mydb")
await db.connect()

# Stream with custom chunk size
async for row in db.query_stream(
    "SELECT * FROM metrics WHERE date > :start_date",
    {"start_date": "2024-01-01"},
    chunk_size=500  # Fetch 500 rows at a time
):
    await process_metric(row)

MongoDB Streaming

from database import MongoDatabase

db = MongoDatabase(database="mydb")
await db.connect()

# Stream documents with query
async for doc in db.query_stream(
    "large_collection",
    {"status": "active"},
    {"sort": [("created_at", -1)]},
    chunk_size=100
):
    await process_document(doc)

Benefits

• Memory Efficient: Only loads chunks into memory, not entire dataset
• Faster Start: Begin processing immediately without waiting for full query
• Scalable: Handle datasets of any size without memory issues
• Configurable: Adjust chunk size for optimal performance

Use Cases

• Processing millions of records
• ETL operations on large datasets
• Real-time data streaming
• Memory-constrained environments
• Progressive data export/import

Connection Pooling

Glean Database includes built-in connection pooling for PostgreSQL, TimescaleDB, and MongoDB to optimize performance and resource usage.

PostgreSQL & TimescaleDB

from database import PostgresDatabase

db = PostgresDatabase(
    database="mydb",
    host="localhost",
    username="postgres",
    password="secret",
    min_pool_size=10,      # Minimum connections (default: 10)
    max_pool_size=20,      # Maximum connections (default: 10)
    max_queries=50000,     # Max queries per connection (default: 50000)
    max_inactive_connection_lifetime=300.0  # Max idle time in seconds (default: 300)
)

await db.connect()  # Creates connection pool
# Pool automatically manages connections
await db.disconnect()  # Closes pool

MongoDB

from database import MongoDatabase

db = MongoDatabase(
    database="mydb",
    host="localhost",
    username="admin",
    password="secret",
    max_pool_size=100,     # Maximum connections (default: 100)
    min_pool_size=10,      # Minimum connections (default: 10)
    max_idle_time_ms=300000  # Max idle time in milliseconds (default: 300000)
)

await db.connect()  # Creates connection pool

Configuration via JSON

{
  "database": "mydb",
  "host": "localhost",
  "port": 5432,
  "username": "postgres",
  "password": "secret",
  "pool": {
    "min_size": 10,
    "max_size": 20,
    "max_queries": 50000,
    "max_inactive_lifetime": 300.0
  }
}

Benefits

• Reduced Overhead: Reuse existing connections instead of creating new ones
• Better Performance: Faster query execution with warm connections
• Resource Management: Automatic cleanup of idle connections
• Scalability: Handle more concurrent operations efficiently

Architecture

Core Components

database/
├── core/
│   ├── database.py      # Abstract Database base class
│   └── cache.py         # Abstract Cache base class
├── backends/
│   ├── sqlite.py        # SQLite implementation
│   ├── postgres.py      # PostgreSQL implementation
│   ├── mongo.py         # MongoDB implementation
│   └── timescale.py     # TimescaleDB implementation
└── cache/
    ├── memory.py        # In-memory cache
    └── redis.py         # Redis cache

Design Principles

1. Abstraction: Common Database and Cache base classes define the interface
2. Composition: Database backends can optionally use cache backends
3. Async-First: All I/O operations are async for high concurrency
4. Type Safety: Full type hints throughout the codebase
5. Testability: All external dependencies are easily mockable

API Reference

Database Base Class

All database backends implement these methods:

async def connect() -> bool

Establish connection to the database.

async def disconnect() -> bool

Close connection to the database.

async def query(query: str, params: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]

Execute a query and return results.

async def execute(statement: str, params: Optional[Dict[str, Any]] = None) -> int

Execute a statement that modifies the database. Returns number of affected rows.

property is_connected -> bool

Check if database is currently connected.

TimescaleDB Extensions

TimescaleDB backend includes additional methods:

async def create_hypertable(...)

Convert a regular table to a TimescaleDB hypertable with automatic time-based partitioning.

async def add_retention_policy(...)

Automatically drop old data based on time intervals.

async def add_compression_policy(...)

Compress older chunks to save storage space.

async def time_bucket_query(...)

Execute time-bucket aggregation queries.

async def continuous_aggregate(...)

Create materialized views that automatically update.

Cache Base Class

All cache backends implement:

async def get(key: str) -> Optional[Any]

Retrieve a value from cache.

async def set(key: str, value: Any, ttl: Optional[timedelta] = None) -> bool

Store a value in cache with optional TTL.

async def delete(key: str) -> bool

Remove a value from cache.

async def clear() -> bool

Clear all cached values.

InMemoryCache Specific

__init__(cleanup_interval: int = 60, max_size: Optional[int] = None)

Initialize in-memory cache with optional configuration.

Parameters:

• cleanup_interval: Seconds between background cleanup runs (default: 60)
• max_size: Maximum number of keys allowed (default: None for unlimited)

start_cleanup()

Start the background task that periodically removes expired keys.

async def stop_cleanup()

Stop the background cleanup task.

Command Line Interface

Glean Database includes a CLI tool for interacting with databases from the command line.

Installation

After installing the package, the glean-db command will be available:

pip install glean-database

Configuration

Create a JSON configuration file with your database connection parameters:

# Show example configurations
python -m database show-config-example

Example config.json:

{
  "database": "./my_database.db",
  "cache": {
    "type": "memory",
    "cleanup_interval": 60,
    "max_size": 1000
  }
}

Commands

Test Connection

python -m database test-connection -c config.json -t sqlite

Execute Queries

# Query with table output (default)
python -m database query -c config.json -t sqlite -q "SELECT * FROM users"

# Query with JSON output
python -m database query -c config.json -t sqlite -q "SELECT * FROM users" -o json

# Query with CSV output
python -m database query -c config.json -t sqlite -q "SELECT * FROM users" -o csv

# Query from file
python -m database query -c config.json -t postgres -f query.sql

# Query with parameters
python -m database query -c config.json -t postgres \
  -q "SELECT * FROM users WHERE id = :id" \
  -p '{"id": 1}'

Execute Statements

# Create table
python -m database execute -c config.json -t sqlite \
  -s "CREATE TABLE users (id INT, name TEXT, email TEXT)"

# Insert data
python -m database execute -c config.json -t sqlite \
  -s "INSERT INTO users VALUES (1, 'John', '[e-mail]')"

# Execute from file
python -m database execute -c config.json -t postgres -f schema.sql

# With parameters
python -m database execute -c config.json -t postgres \
  -s "INSERT INTO users (name, email) VALUES (:name, :email)" \
  -p '{"name": "Jane", "email": "[e-mail]"}'

Initialize Schema from SQL Files

# Initialize database schema from directory of SQL files
python -m database init-schema -c config.json -t postgres -d ~/git/glean/shared/sql/postgres/

# Dry run to validate SQL files without executing
python -m database init-schema -c config.json -t postgres -d ~/sql/ --dry-run

The init-schema command:

• Reads all .sql files from the directory in alphabetical order
• Parses CREATE TABLE, CREATE INDEX, CREATE FUNCTION, and other DDL statements
• Automatically skips tables that already exist (prefers modification over recreation)
• Handles multi-line statements, comments, and PostgreSQL functions
• Provides detailed execution results

CLI Options

Commands:
  query             Execute a query against the database
  execute           Execute a statement that modifies the database  
  init-schema       Initialize database schema from SQL files
  test-connection   Test database connection
  list-backends     List available database backends
  show-config-example  Show example configuration file format

Options:
  -c, --config PATH              Path to JSON configuration file [required]
  -t, --type [sqlite|postgres|mongo|timescale|mysql|mssql|oracle]  Database type [required]
  -q, --query TEXT              SQL/query string to execute
  -s, --statement TEXT          SQL/command to execute
  -f, --file PATH               File containing query/statement
  -d, --directory PATH          Directory containing .sql schema files
  -p, --params TEXT             JSON string with parameters
  -o, --output [table|json|csv] Output format (default: table)
  --dry-run                     Parse SQL files without executing
  --version                     Show version
  --help                        Show help message

Testing

# Run all tests
pytest

# Run with coverage
pytest --cov=database --cov-report=term-missing

# Run specific test file
pytest tests/test_sqlite.py -v

# Run tests for specific database (requires running server)
pytest -m postgres
pytest -m mongo
pytest -m redis
pytest -m timescale

All unit tests use mocks and complete in under 1 second. Integration tests require running database servers.

Development

Setup

# Clone repository
git clone <repo-url>
cd glean-database

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # or `.venv\Scripts\activate` on Windows

# Install dependencies
pip install -e ".[dev]"

Code Style

# Format code
black database/ tests/

# Sort imports
isort database/ tests/

# Type checking
mypy database/

Running Tests

See Testing section above.

Requirements

• Python 3.12+
• aiosqlite >= 0.19.0
• asyncpg >= 0.28.0
• motor >= 3.3.0
• redis[hiredis] >= 5.0.0

License

MIT

Contributing

Contributions are welcome! Please refer to .github/copilot-instructions.md for development guidelines including:

• Code style standards
• Testing requirements (unit tests must complete in < 1 second)
• Documentation standards
• Error handling patterns

Schema Management

Glean Database includes powerful schema management capabilities for initializing and maintaining database schemas from SQL files.

Features

• Directory-based: Organize schema files in a directory (e.g., 01_tables.sql, 02_indexes.sql)
• Automatic ordering: Files are processed in alphabetical order
• Idempotent: Safely re-run schema initialization - existing tables are skipped
• Comprehensive parsing: Handles CREATE TABLE, CREATE INDEX, CREATE FUNCTION, triggers, views, and more
• Comment support: Strips SQL comments (single-line -- and multi-line /* */)
• Error handling: Continues processing even if individual statements fail
• Dry-run mode: Validate SQL files without making changes

Usage

from database import PostgresDatabase
from database.core.schema import read_sql_files, apply_schema

async def initialize_database():
    db = PostgresDatabase(database="mydb", host="localhost")
    await db.connect()
    
    # Read SQL files from directory
    sql_files = read_sql_files("~/sql/postgres/")
    
    # Apply schema
    results = await apply_schema(db, sql_files)
    
    print(f"Successful: {results['successful']}")
    print(f"Failed: {results['failed']}")
    print(f"Skipped: {results['skipped']}")
    
    await db.disconnect()

CLI Usage

# Initialize schema from SQL files
python -m database init-schema \
  -c config.json \
  -t postgres \
  -d ~/git/glean/shared/sql/postgres/

# Dry run to validate
python -m database init-schema \
  -c config.json \
  -t postgres \
  -d ~/sql/ \
  --dry-run

Example Directory Structure

sql/postgres/
├── 01_mcp_tables.sql      # Core tables
├── 02_trading_tables.sql  # Domain-specific tables
├── 03_indexes.sql         # Indexes for performance
├── 04_functions.sql       # Stored procedures
└── 05_triggers.sql        # Triggers and automation

Backend Support

The table_exists() method is implemented for:

• ✓ PostgreSQL
• ✓ SQLite
• ✓ MySQL
• ✓ MSSQL
• ✓ Oracle
• ✓ TimescaleDB (inherits from PostgreSQL)

Roadmap

• CLI tools for database management
• Connection pooling configuration
• Query result streaming for large datasets
• MySQL support
• MSSQL (SQL Server) support
• Oracle support
• Schema management from SQL files
• MariaDB support (use MySQL driver)
• Cassandra support
• Transaction support
• Schema migration utilities (ALTER TABLE support)
• Interactive shell (REPL)
• Schema introspection and visualization