dd-db 0.1.0

Unified Relational DB abstraction layer — clean adapters for 9+ databases

dd-db

Unified Relational DB abstraction layer for Python.

Connect to any relational database and get a pandas DataFrame back — with a consistent API for schema inspection, connection management, and query timing.

Supported Databases

Adapter	Class	Extra
SQLite (stdlib)	SQLiteDB	(none)
DuckDB	DuckDB	duckdb
PostgreSQL	PostgresDB	postgres
MySQL / MariaDB	MySQLDB	mysql
Snowflake	SnowflakeDB	snowflake
Google BigQuery	BigQueryDB	bigquery
ClickHouse	ClickHouseDB	clickhouse
SQL Server	MSSQLDB	mssql
Oracle	OracleDB	oracle

Install

pip install dd-db                   # SQLite only (stdlib, zero extra deps)
pip install "dd-db[duckdb]"         # + DuckDB
pip install "dd-db[postgres]"       # + PostgreSQL
pip install "dd-db[all]"            # all adapters
pip install "dd-db[dev]"            # dev tools + DuckDB

Quick Start

from dd_db import SQLiteDB

with SQLiteDB(":memory:") as db:
    db.run_query("CREATE TABLE t (id INT, name TEXT)")
    db.run_query("INSERT INTO t VALUES (1, 'Alice')")
    db.run_query("INSERT INTO t VALUES (2, 'Bob')")

    # SELECT returns a pandas DataFrame
    df = db.run_query("SELECT * FROM t")
    print(df)
    #    id   name
    # 0   1  Alice
    # 1   2    Bob

    # Parameterised queries
    row = db.run_query("SELECT * FROM t WHERE id = :id", params={"id": 1})

    # Schema inspection
    print(db.list_tables())          # ['t']
    print(db.describe("t"))          # DataFrame with column/type/pk columns
    schema = db.get_schema("t")      # TableSchema Pydantic model
    print(schema.row_count)          # 2

    # Timed query
    df, meta = db.timed_query("SELECT * FROM t")
    print(meta.execution_time_ms)    # e.g. 0.42

API Reference

Core methods (all adapters)

Method	Returns	Description
run_query(sql, params?)	DataFrame	Execute SQL; SELECT → rows, DML → {rows_affected}
list_tables(schema?)	list[str]	Table names
tables(schema?)	DataFrame	Table names as DataFrame
get_schema(table, schema?)	TableSchema	Typed column metadata
describe(table, schema?)	DataFrame	Human-readable column info
test_connection()	bool	Health check
connection_info()	ConnectionInfo	Loggable summary (no passwords)
timed_query(sql, params?)	(DataFrame, QueryResult)	Query + execution metadata
connect()	None	Open connection
disconnect()	None	Close connection

Context manager

with SQLiteDB("mydb.sqlite") as db:
    ...
# connection closed automatically

Pydantic models

from dd_db import TableSchema, ColumnInfo, QueryResult, ConnectionInfo

• ColumnInfo — name, data_type, nullable, default, primary_key
• TableSchema — table_name, schema_name, columns, row_count, full_name
• QueryResult — sql, rows_returned, columns, execution_time_ms, success, error
• ConnectionInfo — adapter, host, port, database, username

Examples

DuckDB analytics

from dd_db import DuckDB

with DuckDB() as db:
    df = db.run_query("""
        SELECT region, SUM(amount) AS total
        FROM sales
        GROUP BY region
        ORDER BY total DESC
    """)
    print(df)

PostgreSQL

from dd_db import PostgresDB

with PostgresDB(host="localhost", database="mydb", user="me", password="pw") as db:
    df = db.run_query("SELECT * FROM orders WHERE status = :status",
                      params={"status": "pending"})

Snowflake

from dd_db import SnowflakeDB

with SnowflakeDB(account="xy12345", user="me", password="pw",
                 database="ANALYTICS", schema="PUBLIC",
                 warehouse="COMPUTE_WH") as db:
    df = db.run_query("SELECT TOP 100 * FROM my_table")

Cookbooks

See cookbook/ for runnable examples:

• 01_sqlite_basics.py — full walkthrough with SQLite (no install needed)
• 02_duckdb_basics.py — analytics with DuckDB, including DataFrame joins

Running Tests

pip install -e ".[dev]"
python -m pytest

Tests use :memory: SQLite — no external server required.

Design

See docs/DESIGN.md for:

• Why synchronous-first?
• Why DataFrame return type everywhere?
• Relationship to vanna.ai
• How to add a new adapter

License

MIT