The Python counterpart to pgsql-test on npm. Instant, isolated PostgreSQL databases for each test — with automatic transaction rollbacks, context switching, and clean seeding.
New to pgpm? Check out the Workspace Setup Guide for a complete walkthrough of creating a pgpm workspace with Python tests.
- Instant test DBs — each one seeded, isolated, and UUID-named
- Per-test rollback — every test runs in its own transaction with savepoint-based rollback via
before_each()/after_each() - RLS-friendly — test with role-based auth via
set_context() - pgpm integration — run database migrations using pgpm (PostgreSQL Package Manager)
- Flexible seeding — run
.sqlfiles, programmatic seeds, pgpm modules, or combine multiple strategies - Auto teardown — no residue, no reboots, just clean exits
# Using Poetry (recommended)
poetry add pgsql-test
# Using pip
pip install pgsql-testimport pytest
from pgsql_test import get_connections, seed
# Basic usage
def test_basic_query():
conn = get_connections()
result = conn.db.query('SELECT 1 as value')
assert result.rows[0]['value'] == 1
conn.teardown()
# With pytest fixture
@pytest.fixture
def db():
conn = get_connections()
yield conn.db
conn.teardown()
def test_with_fixture(db):
result = db.query('SELECT 1 as value')
assert result.rows[0]['value'] == 1The primary use case for pgsql-test is testing PostgreSQL modules managed by pgpm. The seed.pgpm() adapter runs pgpm deploy to apply your migrations to an isolated test database.
Install pgpm globally:
npm install -g pgpmimport pytest
from pgsql_test import get_connections, seed
@pytest.fixture
def db():
conn = get_connections(
seed_adapters=[
seed.pgpm(
module_path="./packages/my-module",
package="my-module"
)
]
)
db = conn.db
db.before_each()
yield db
db.after_each()
conn.teardown()
def test_my_function(db):
# Your pgpm module's functions are now available
result = db.one("SELECT my_schema.my_function() as result")
assert result['result'] == expected_valueIf your module depends on other pgpm packages (like @pgpm/faker), install them first:
cd packages/my-module
pgpm install @pgpm/fakerThen test:
def test_faker_integration(db):
# @pgpm/faker functions are available after pgpm deploy
result = db.one("SELECT faker.city('MI') as city")
assert result['city'] is not NoneA typical pgpm workspace for testing looks like:
my-workspace/
pgpm.json # Workspace config
packages/
my-module/
package.json # Module metadata
my-module.control # PostgreSQL extension control
pgpm.plan # Migration plan
deploy/
schemas/
my_schema.sql # CREATE SCHEMA my_schema;
functions/
my_function.sql # CREATE FUNCTION ...
revert/
schemas/
my_schema.sql # DROP SCHEMA my_schema;
verify/
schemas/
my_schema.sql # SELECT 1 FROM ...
| Parameter | Type | Description |
|---|---|---|
module_path |
str |
Path to the pgpm module directory |
package |
str |
Package name to deploy (required to avoid interactive prompts) |
deploy_args |
list[str] |
Additional arguments to pass to pgpm deploy |
cache |
bool |
Enable caching (not yet implemented) |
For simpler use cases without pgpm, seed directly from SQL files:
@pytest.fixture
def seeded_db():
conn = get_connections(
seed_adapters=[seed.sqlfile(['schema.sql', 'fixtures.sql'])]
)
yield conn.db
conn.teardown()
def test_with_seeding(seeded_db):
users = seeded_db.many('SELECT * FROM users')
assert len(users) > 0The before_each() and after_each() methods provide automatic transaction rollback for each test. This ensures complete isolation between tests - any changes made during a test are automatically rolled back, so each test starts with a clean slate.
before_each()begins a transaction and creates a savepoint- Your test runs and makes changes to the database
after_each()rolls back to the savepoint, undoing all changes- The next test starts fresh with only the seeded data
@pytest.fixture
def db():
conn = get_connections(
seed_adapters=[seed.sqlfile(['schema.sql'])]
)
db = conn.db
db.before_each() # Begin transaction + savepoint
yield db
db.after_each() # Rollback to savepoint
conn.teardown()
def test_insert_user(db):
# This insert will be rolled back after the test
db.execute("INSERT INTO users (name) VALUES ('Test User')")
result = db.one("SELECT * FROM users WHERE name = 'Test User'")
assert result['name'] == 'Test User'
def test_user_count(db):
# Previous test's insert is not visible here
result = db.one("SELECT COUNT(*) as count FROM users")
assert result['count'] == 0 # Only seeded dataWithout per-test rollback, tests can interfere with each other:
- Test A inserts a user
- Test B expects 0 users but finds 1
- Tests become order-dependent and flaky
With before_each()/after_each(), each test is completely isolated, making your test suite reliable and deterministic.
Test Row Level Security policies by switching contexts:
def test_rls_policy(db):
db.before_each()
# Set the user context
db.set_context({'app.user_id': '123'})
# Now queries will be filtered by RLS policies
result = db.many('SELECT * FROM user_data')
db.after_each()seed.pgpm(module_path="./packages/my-module", package="my-module")seed.sqlfile(['schema.sql', 'fixtures.sql'])seed.fn(lambda ctx: ctx['pg'].execute(
"INSERT INTO users (name) VALUES (%s)", ('Alice',)
))seed.compose([
seed.pgpm(module_path="./packages/my-module", package="my-module"),
seed.sqlfile(['fixtures.sql']),
seed.fn(lambda ctx: ctx['pg'].execute("INSERT INTO ...")),
])Configure via environment variables:
export PGHOST=localhost
export PGPORT=5432
export PGUSER=postgres
export PGPASSWORD=your_passwordOr pass configuration directly:
conn = get_connections(
pg_config={
'host': 'localhost',
'port': 5432,
'user': 'postgres',
'password': 'your_password',
}
)Creates a new isolated test database and returns connection objects.
Returns a ConnectionResult with:
pg: PgTestClient connected as superuserdb: PgTestClient for testing (same as pg for now)admin: DbAdmin for database managementmanager: PgTestConnector managing connectionsteardown(): Function to clean up
query(sql, params?): Execute SQL and return QueryResultone(sql, params?): Return exactly one rowone_or_none(sql, params?): Return one row or Nonemany(sql, params?): Return multiple rowsmany_or_none(sql, params?): Return rows (may be empty)execute(sql, params?): Execute and return affected row countbefore_each(): Start test isolation (transaction + savepoint)after_each(): End test isolation (rollback)set_context(dict): Set session variables for RLS testing
Here's a complete CI workflow for testing pgpm modules:
name: Test
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:17
env:
POSTGRES_USER: postgres
POSTGRES_PASSWORD: password
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
env:
PGHOST: localhost
PGPORT: 5432
PGUSER: postgres
PGPASSWORD: password
steps:
- uses: actions/checkout@v4
- uses: actions/setup-node@v4
with:
node-version: '20'
- name: Install pgpm
run: npm install -g pgpm
- uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install Poetry
uses: snok/install-poetry@v1
- name: Install dependencies
run: poetry install
- name: Bootstrap pgpm roles
run: |
pgpm admin-users bootstrap --yes
pgpm admin-users add --test --yes
- name: Run tests
run: poetry run pytest -v# Install dependencies
poetry install
# Run tests
poetry run pytest
# Run linting
poetry run ruff check .
# Run type checking
poetry run mypy src- pgsql-test - The original TypeScript/Node.js version
- pgpm - PostgreSQL Package Manager
MIT
🛠Built by the Constructive team — creators of modular Postgres tooling for secure, composable backends. If you like our work, contribute on GitHub.
AS DESCRIBED IN THE LICENSES, THE SOFTWARE IS PROVIDED "AS IS", AT YOUR OWN RISK, AND WITHOUT WARRANTIES OF ANY KIND.
No developer or entity involved in creating this software will be liable for any claims or damages whatsoever associated with your use, inability to use, or your interaction with other users of the code, including any direct, indirect, incidental, special, exemplary, punitive or consequential damages, or loss of profits, cryptocurrencies, tokens, or anything else of value.