A PostgreSQL async library for modular applications

pgdbm allows you to build database-backed libraries that work in any deployment context.

Most DB tools assume the application owns the database. pgdbm is designed for building libraries that have their own tables and migrations, and that work identically whether running standalone as a service or embedded in a larger application sharing a connection pool.

Built on asyncpg, it gives you:

  • Connection pooling with retries and backoff
  • Schema-aware SQL templating ({{tables.users}}) for portable queries
  • A migration system with per-module isolation and checksums
  • Transaction management with savepoint support
  • Optional monitoring (slow queries, pool stats, EXPLAIN)
  • Testing utilities (ephemeral DBs, fixtures, isolation)
  • A CLI for migrations and database management

xkcd Exploits of a Mom

https://github.com/juanre/pgdbm

Installation

uv add pgdbm

For CLI support:

uv add "pgdbm[cli]"

Core ideas

  • SQL-centric: no ORM, just fast asyncpg with safe parameter binding
  • {{tables.name}} expands to schema.name at query time, making code portable
  • Migrations are module-scoped with checksums and advisory-locked application
  • Production guardrails: TLS, statement/idle/lock timeouts

Quick usage

Standalone (module owns the DB)

from pgdbm import AsyncDatabaseManager, DatabaseConfig, AsyncMigrationManager

config = DatabaseConfig(
    connection_string="postgresql://localhost/app",
    schema="users",
    # Production guardrails
    ssl_enabled=False,                    # Enable with 'verify-full' in production
    statement_timeout_ms=60_000,
    idle_in_transaction_session_timeout_ms=60_000,
    lock_timeout_ms=5_000,
)

db = AsyncDatabaseManager(config)
await db.connect()

migrations = AsyncMigrationManager(db, migrations_path="./migrations", module_name="users")
await migrations.apply_pending_migrations()

user_id = await db.fetch_value(
    "INSERT INTO {{tables.users}} (email) VALUES ($1) RETURNING id",
    "[email protected]",
)

Shared pool (module uses a DB owned by another)

# The host application owns the pool
shared_pool = await AsyncDatabaseManager.create_shared_pool(
    DatabaseConfig(connection_string="postgresql://localhost/app")
)

# Your module uses the pool but owns its schema and migrations
user_db = AsyncDatabaseManager(pool=shared_pool, schema="users")
await AsyncMigrationManager(user_db, "./migrations", module_name="users").apply_pending_migrations()

Migrations

-- migrations/001_initial.sql
CREATE TABLE IF NOT EXISTS {{tables.users}} (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
  • Use {{tables.name}} in DDL/DML; it is expanded safely at runtime
  • Migrations are tracked in schema_migrations with checksums
  • Set a unique module_name per module/library

Transactions

async with db.transaction() as tx:
    await tx.execute("INSERT INTO {{tables.users}} (email) VALUES ($1)", "[email protected]")
    await tx.execute("INSERT INTO {{tables.logs}} (msg) VALUES ($1)", "User created")
    # Commits on success, rolls back on exception

    # Nested transactions use savepoints
    async with tx.transaction() as nested:
        await nested.execute("UPDATE {{tables.users}} SET active = true")

Monitoring (optional)

from pgdbm import MonitoredAsyncDatabaseManager

db = MonitoredAsyncDatabaseManager(config)
await db.connect()
metrics = await db.get_metrics()          # pool sizes, query stats
slow = db.get_slow_queries(200)           # queries over 200ms
plan = await db.explain_query("SELECT * FROM {{tables.users}} WHERE id = $1", 42)

Testing utilities

# conftest.py
from pgdbm.fixtures.conftest import *

# test_users.py
async def test_user_flow(test_db):
    await test_db.execute("CREATE TABLE {{tables.users}} (id serial primary key, email text)")
    uid = await test_db.fetch_value(
        "INSERT INTO {{tables.users}} (email) VALUES ($1) RETURNING id", "[email protected]"
    )
    assert uid == 1

Production notes

  • Enable TLS with certificate verification:
    • ssl_enabled=True, ssl_mode='verify-full', and ssl_ca_file as needed
  • Use server-side timeouts:
    • statement_timeout_ms, idle_in_transaction_session_timeout_ms, lock_timeout_ms
  • Migrations are serialized per module_name via advisory locks
  • Keep debug logging off in production (args are masked, but prefer redaction)

When to use pgdbm

  • You’re building a library or service that needs its own database tables
  • That library must work both standalone and embedded in larger applications
  • You want SQL and asyncpg performance without an ORM
  • You need per-module migrations with safe schema isolation