pgfast 0.7.1
Lightweight PostgreSQL migration and testing toolkit for FastAPI with raw SQL, dependency management, and isolated test databases
Requires Python: >=3.14

# pgfast
[![codecov](https://codecov.io/gh/elmcrest/pgfast/graph/badge.svg?token=dFtlfgtEQx)](https://codecov.io/gh/elmcrest/pgfast)
![CI](https://github.com/elmcrest/pgfast/actions/workflows/ci_cd.yml/badge.svg)
[![PyPI version](https://img.shields.io/pypi/v/pgfast)](https://pypi.org/project/pgfast/)
[![Python versions](https://img.shields.io/pypi/pyversions/pgfast)](https://pypi.org/project/pgfast/)
[![License](https://img.shields.io/pypi/l/pgfast)](https://pypi.org/project/pgfast/)

**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.

## Why pgfast?

- **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.

## Positioning

### Who it's for

- 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).

### Who it's not for

- 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`).

### How it compares

| Option | What it covers | Testing story | Tradeoffs |
| --- | --- | --- | --- |
| **pgfast** | `asyncpg` pooling + FastAPI integration + SQL-file migrations | DB-per-test fixtures with template cloning | You own the SQL; fewer abstractions |
| **Alembic** | Migrations (commonly with SQLAlchemy) | Up to you (often custom) | Not a runtime DB layer; more moving parts |
| **SQLAlchemy ORM (+ Alembic)** | ORM + query building + migrations | Up to you | More abstraction; less “just SQL” |
| **Testcontainers (Postgres)** | Hermetic Postgres for tests/CI | Very isolated; container startup cost | Doesn’t provide migrations or runtime access by itself |

## Installation

```bash
pip install pgfast
```

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

## Quick Start

### 1. Set up your FastAPI app

```python
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")
```

### 2. Create and run migrations

```bash
# 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
```

### 3. Write tests with isolated databases

```python
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', 'alice@example.com')
        """)

        user = await conn.fetchrow("SELECT * FROM users WHERE name = 'Alice'")
        assert user["email"] == "alice@example.com"
```

## Features

### Connection Management
- 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

### Schema Migrations
- 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

### Test Database Management
- 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

### CLI Commands

```bash
# 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
```

## Migration Features

### Dependency Tracking

Declare dependencies between migrations using comments:

```sql
-- 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.

### Checksum Validation

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

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

### Dry-Run Mode

Preview migrations before applying them:

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

### Long-Running Migrations

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:

```bash
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).

### Organizing Migrations at Scale

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/`

## Testing

pgfast includes pytest fixtures for fast, isolated testing:

```python
# 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:
```bash
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.

### Advanced Testing

#### Selective Fixture Loading

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

```python
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:
        # ...
```

#### Fixture Reusability

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`.

#### Multiple Databases

Need to test cross-database interactions? Use `db_pool_factory`:

```python
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)
```

## Row-Level Security (RLS)

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.

### Basic Usage

```python
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")
```

### Multiple Session Variables

Pass multiple settings for complex authorization:

```python
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)
```

### Example RLS Policy

```sql
-- 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);
```

### PgBouncer Compatibility

`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.

### Transaction Context

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.

## Configuration

```python
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)

## Philosophy

**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.

## Development

```bash
# 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
```

## License

MIT

---

Built with [asyncpg](https://github.com/MagicStack/asyncpg) and [FastAPI](https://fastapi.tiangolo.com/).