Lightweight asyncpg integration for FastAPI. Raw SQL. Fast tests. Zero magic.

pgfast gives you everything you need to build FastAPI applications with PostgreSQL connection pooling, migrations, and isolated test databases without the weight of an ORM. Write SQL, own your queries, ship faster.

• Raw SQL: Write the queries you want. No ORM translation layer, no query builder abstraction.
• Fast Tests: Template database cloning gives you isolated test databases in milliseconds, not seconds.
• FastAPI Native: Lifespan integration and dependency injection that feels natural.
• Simple Migrations: Timestamped SQL files. Up and down. That's it.
• Built for Testing: Pytest fixtures included. Create isolated databases, load fixtures, test in parallel.

• FastAPI + PostgreSQL projects using asyncpg that want a small, explicit database layer.
• Teams that prefer writing SQL directly and using PostgreSQL features (RLS, triggers, CTEs, etc.).
• Codebases where integration tests need to be fast, isolated, and parallel-friendly (database per test).

• You want an ORM/unit-of-work pattern, model relationships, or query composition (use SQLAlchemy ORM).
• You want migration autogeneration from models or a more full-featured migration workflow (use Alembic).
• You're not on PostgreSQL (pgfast is Postgres-specific via asyncpg).

pip install pgfast

Requires Python 3.14+ and PostgreSQL (earlier versions should work, open a PR if you'd like to add support).

from fastapi import FastAPI, Depends
from pgfast import DatabaseConfig, create_lifespan, get_db_pool
import asyncpg

config = DatabaseConfig(url="postgresql://localhost/mydb")
app = FastAPI(lifespan=create_lifespan(config))

@app.get("/users")
async def get_users(pool: asyncpg.Pool = Depends(get_db_pool)):
    async with pool.acquire() as conn:
        return await conn.fetch("SELECT id, name FROM users")

# Create a migration
pgfast schema create your/module add_users_table

# Edit the generated SQL files in your/module/migrations/
# Then preview and apply migrations
export DATABASE_URL="postgresql://localhost/mydb"
pgfast schema up --dry-run  # Preview first
pgfast schema up            # Apply migrations

import pytest
from pgfast.pytest import isolated_db

async def test_user_creation(isolated_db):
    """Each test gets a fresh database fast and isolated."""
    async with isolated_db.acquire() as conn:
        await conn.execute("""
            INSERT INTO users (name, email)
            VALUES ('Alice', '[email protected]')
        """)

        user = await conn.fetchrow("SELECT * FROM users WHERE name = 'Alice'")
        assert user["email"] == "[email protected]"

• asyncpg connection pooling with configurable size and timeouts
• Graceful startup and shutdown with FastAPI lifespan
• Connection validation on pool creation
• RLS-aware connections with per-request session variables

• Timestamped migration files: {timestamp}_{name}_up.sql and _down.sql
• Transactional migration application
• CLI for creating, applying, and rolling back migrations
• Migration status tracking
• Dependency tracking: Declare dependencies between migrations
• Checksum validation: Detect modified migrations automatically
• Dry-run mode: Preview changes before applying

• Isolated test databases for every test
• Template database cloning for ~10-100x faster test setup (*needs benchmarking)
• Automatic cleanup
• Fixture loading from SQL files
• Pytest fixtures ready to use

# Migration Management
pgfast schema create <module_path> <name>    # Create migration files
pgfast schema up                             # Apply pending migrations
pgfast schema up --target <version>          # Migrate to specific version
pgfast schema up --dry-run                   # Preview migrations without applying
pgfast schema up --force                     # Skip checksum validation
pgfast schema up --timeout <seconds>         # Set query timeout (default: no limit)
pgfast schema down --steps 1                 # Rollback 1 migration
pgfast schema down --target <version>        # Rollback to specific version
pgfast schema down --dry-run                 # Preview rollback
pgfast schema down --timeout <seconds>       # Set query timeout (default: no limit)
pgfast schema status                         # Show migration status
pgfast schema deps                           # Show dependency graph
pgfast schema verify                         # Verify migration checksums

# Test Database Management
pgfast test-db create                        # Create test database
pgfast test-db list                          # List test databases
pgfast test-db cleanup                       # Clean up test databases

Declare dependencies between migrations using comments:

-- depends_on: 20240101000000, 20240102000000

CREATE TABLE posts (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    title VARCHAR(255)
);

Migrations are automatically applied in dependency order, and circular dependencies are detected.

Migrations are checksummed (SHA-256) when applied. pgfast automatically detects if migration files have been modified after being applied:

pgfast schema verify  # Check for modifications
pgfast schema up      # Validates checksums automatically
pgfast schema up --force  # Skip validation if needed

Preview migrations before applying them:

pgfast schema up --dry-run    # See what would be applied
pgfast schema down --dry-run  # See what would be rolled back

By default, migrations run without a query timeout, allowing complex migrations (large data migrations, index creation, etc.) to complete without interruption. If you need to set a specific timeout:

pgfast schema up --timeout 600    # 10 minute timeout
pgfast schema down --timeout 300  # 5 minute timeout

This overrides the default command_timeout from the connection pool configuration, which is typically set for regular queries (default: 60 seconds).

As your project grows, organize migrations using subdirectories. pgfast automatically discovers migrations in nested directories:

db/migrations/
├── users/
│   ├── 20250101000000_create_users_table_up.sql
│   ├── 20250101000000_create_users_table_down.sql
│   └── 20250305120000_add_email_verification_up.sql
├── products/
│   └── 20250102000000_create_products_table_up.sql
├── orders/
│   └── 20250103000000_create_orders_table_up.sql
└── auth/
    └── 20250115000000_add_oauth_providers_up.sql

Subdirectories are discovered automatically via the **/migrations pattern. Dependencies work across subdirectories, and migrations are applied in timestamp order regardless of directory structure.

You can organize by:

• Domain/Feature: users/, products/, orders/
• Release: v1.0/, v1.1/, v2.0/
• Date + Domain: 2025-01-auth/, 2025-02-products/

pgfast includes pytest fixtures for fast, isolated testing:

# tests/conftest.py
from pgfast.pytest import *

# Your tests automatically get:
# - isolated_db: Fresh database per test (with template optimization)
# - db_pool_factory: Create multiple databases in one test
# - db_with_fixtures: Database with fixtures pre-loaded

Run tests:

export TEST_DATABASE_URL="postgresql://localhost/postgres"
pytest              # Run tests sequentially
pytest -n auto      # Run tests in parallel (recommended)

The test infrastructure supports parallel execution out of the box. Each test gets an isolated database, so tests can run concurrently without conflicts.

Instead of loading all fixtures with db_with_fixtures, you can load specific fixtures using the fixture_loader fixture:

async def test_specific_feature(isolated_db, fixture_loader):
    # Load only the 'users' and 'products' fixtures
    # This will automatically load them in dependency order
    await fixture_loader(["users", "products"])
    
    async with isolated_db.acquire() as conn:
        # ...

Fixtures are defined as SQL files following the naming convention {version}_{name}_fixture.sql. pgfast automatically discovers fixtures across multiple directories (e.g., db/fixtures/, or any directory matching **/fixtures pattern).

• Auto-Discovery: Fixtures are discovered across multiple directories automatically. You can have fixtures in db/fixtures/, module_a/fixtures/, etc.
• Subdirectory Support: Like migrations, fixtures support subdirectories for organization:

  db/fixtures/
  ├── users/
  │   └── 20250101000000_create_users_fixture.sql
  └── products/
      └── 20250102000000_create_products_fixture.sql
  

• Version Matching: The version number MUST match a corresponding migration version.
• Dependency Order: Fixtures are loaded in the same order as their corresponding migrations.
• Reusability: All discovered fixtures are available globally and can be used in any test via fixture_loader or db_with_fixtures.

Need to test cross-database interactions? Use db_pool_factory:

async def test_multi_db(db_pool_factory):
    # Create two isolated databases
    pool1 = await db_pool_factory()
    pool2 = await db_pool_factory()

    try:
        # Test interaction between databases
        pass
    finally:
        # Cleanup is handled automatically, but you can be explicit
        await db_pool_factory.cleanup(pool1)
        await db_pool_factory.cleanup(pool2)

pgfast supports multi-tenant applications using PostgreSQL Row-Level Security. The create_rls_dependency() function creates a FastAPI dependency that sets session variables per-request, which your RLS policies can reference.

from fastapi import FastAPI, Depends, Request
from pgfast import DatabaseConfig, create_lifespan, create_rls_dependency
import asyncpg

config = DatabaseConfig(url="postgresql://localhost/mydb")
app = FastAPI(lifespan=create_lifespan(config))

async def get_tenant_settings(request: Request) -> dict[str, str]:
    # Extract tenant from JWT, header, subdomain, etc.
    tenant_id = request.headers.get("X-Tenant-ID", "")
    return {"app.tenant_id": tenant_id}

get_rls_connection = create_rls_dependency(get_tenant_settings)

@app.get("/items")
async def list_items(conn: asyncpg.Connection = Depends(get_rls_connection)):
    # RLS policies using current_setting('app.tenant_id') filter automatically
    return await conn.fetch("SELECT * FROM items")

Pass multiple settings for complex authorization:

async def get_rls_settings(request: Request) -> dict[str, str]:
    return {
        "app.tenant_id": request.state.tenant_id,
        "app.user_id": request.state.user_id,
        "app.role": request.state.role,
    }

get_rls_connection = create_rls_dependency(get_rls_settings)

-- Migration: Enable RLS and create policy
ALTER TABLE items ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON items
    USING (tenant_id = current_setting('app.tenant_id')::integer);

create_rls_dependency() uses PostgreSQL's set_config() with LOCAL scope, making it safe for connection poolers like PgBouncer in transaction pooling mode. Session variables are transaction-scoped and automatically reset when the request completes - they never leak between clients.

All queries within an RLS dependency execute inside a transaction (required for SET LOCAL semantics). For read-only queries this has no practical impact. For write operations, be aware you're already in a transaction context.

from pgfast import DatabaseConfig

config = DatabaseConfig(
    url="postgresql://localhost/mydb",
    min_connections=5,
    max_connections=20,
    timeout=10.0,
    migrations_dirs=["db/migrations"],
    fixtures_dirs=["db/fixtures"],
)

When using the CLI, you can also use environment variables:

• DATABASE_URL: Connection string
• PGFAST_MIGRATIONS_DIRS: Colon-separated migration directories (overrides auto-discovery)
• PGFAST_FIXTURES_DIRS: Colon-separated fixture directories (overrides auto-discovery)

SQL is not the enemy. Modern PostgreSQL is incredibly powerful. Instead of hiding it behind abstraction layers, pgfast embraces it. Write migrations in SQL. Write queries in SQL. Use PostgreSQL features directly.

Tests should be fast. Creating a database per test shouldn't take seconds. With template database cloning, you get isolation without the wait.

Integration should be simple. No complex configuration, no global state, no magic. Just functions and fixtures that do what they say.

# get a running postgres instance, f.e. with docker:
docker run --rm --name $POSTGRES_DB -e POSTGRES_USER=$POSTGRES_USER -e POSTGRES_PASSWORD=$POSTGRES_PASSWORD -e POSTGRES_DB=$POSTGRES_DB -p ${POSTGRES_PORT:-5432}:5432 postgres:18-alpine

# Run tests
pytest              # Sequential
pytest -n auto      # Parallel (faster)

# Run with coverage
pytest --cov=src/pgfast

# Run integration tests
export TEST_DATABASE_URL="postgresql://localhost/postgres"
pytest tests/integration/
pytest -n auto tests/integration/  # Parallel

MIT

Built with asyncpg and FastAPI.