Postgresql fixtures and fixture factories for Pytest.
npx @tessl/cli install tessl/pypi-pytest-postgresql@7.0.0A pytest plugin providing comprehensive PostgreSQL database fixtures and fixture factories for testing applications that require a real PostgreSQL database instance. The plugin handles the complete lifecycle of PostgreSQL instances during testing including startup, initialization, connection management, and teardown, with built-in support for connection pooling and transaction isolation.
pip install pytest-postgresqlfrom pytest_postgresql import factoriesFor creating custom fixtures:
from pytest_postgresql.factories import postgresql_proc, postgresql_noproc, postgresqlFor accessing utility classes and functions:
from pytest_postgresql.executor import PostgreSQLExecutor
from pytest_postgresql.executor_noop import NoopExecutor
from pytest_postgresql.janitor import DatabaseJanitor
from pytest_postgresql.exceptions import ExecutableMissingException, PostgreSQLUnsupported
from pytest_postgresql.loader import build_loader, sql
from pytest_postgresql.retry import retry
from pytest_postgresql.factories import PortTypeThe plugin automatically provides three main fixtures when installed:
def test_example_postgres(postgresql):
"""Test using the default postgresql fixture."""
cur = postgresql.cursor()
cur.execute("CREATE TABLE test (id serial PRIMARY KEY, num integer, data varchar);")
postgresql.commit()
cur.close()
def test_postgresql_process(postgresql_proc):
"""Test the PostgreSQL process fixture."""
assert postgresql_proc.running() is Truefrom pytest_postgresql import factories
# Create custom process fixture
postgresql_my_proc = factories.postgresql_proc(
port=None,
unixsocketdir='/var/run'
)
# Create custom client fixture using the custom process
postgresql_my = factories.postgresql('postgresql_my_proc')# Using SQL file
postgresql_with_data = factories.postgresql_proc(
load=['/path/to/schema.sql', '/path/to/data.sql']
)
# Using Python function
def load_test_data(**kwargs):
import psycopg
with psycopg.connect(**kwargs) as conn:
with conn.cursor() as cur:
cur.execute("INSERT INTO users (name) VALUES ('test_user')")
conn.commit()
postgresql_with_func = factories.postgresql_proc(
load=[load_test_data]
)The plugin is organized around three main patterns:
This design enables flexible testing scenarios from single-use databases to shared instances across test sessions, with automatic cleanup and isolation between tests.
Creates session-scoped fixtures that start and manage PostgreSQL server processes, with full control over server configuration, data directory management, and initialization.
def postgresql_proc(
executable: Optional[str] = None,
host: Optional[str] = None,
port: Optional[PortType] = -1,
user: Optional[str] = None,
password: Optional[str] = None,
dbname: Optional[str] = None,
options: str = "",
startparams: Optional[str] = None,
unixsocketdir: Optional[str] = None,
postgres_options: Optional[str] = None,
load: Optional[List[Union[Callable, str, Path]]] = None,
) -> Callable[[FixtureRequest, TempPathFactory], Iterator[PostgreSQLExecutor]]: ...Creates fixtures for connecting to existing PostgreSQL servers, ideal for containerized environments, CI systems, or when PostgreSQL is managed externally.
def postgresql_noproc(
host: Optional[str] = None,
port: Union[str, int, None] = None,
user: Optional[str] = None,
password: Optional[str] = None,
dbname: Optional[str] = None,
options: str = "",
load: Optional[List[Union[Callable, str, Path]]] = None,
) -> Callable[[FixtureRequest], Iterator[NoopExecutor]]: ...Creates database connection fixtures with automatic cleanup, transaction isolation, and database management between tests.
def postgresql(
process_fixture_name: str,
dbname: Optional[str] = None,
isolation_level: Optional[psycopg.IsolationLevel] = None,
) -> Callable[[FixtureRequest], Iterator[Connection]]: ...Comprehensive configuration system supporting command-line options, pytest.ini settings, and programmatic configuration for all aspects of PostgreSQL testing.
class PostgresqlConfigDict(TypedDict):
exec: str
host: str
port: Optional[str]
port_search_count: int
user: str
password: str
options: str
startparams: str
unixsocketdir: str
dbname: str
load: List[Union[Path, str]]
postgres_options: str
drop_test_database: bool
def get_config(request: FixtureRequest) -> PostgresqlConfigDict: ...Flexible data loading system supporting SQL files, Python callables, and import strings for database initialization and test data setup.
def build_loader(load: Union[Callable, str, Path]) -> Callable: ...
def sql(sql_filename: Path, **kwargs: Any) -> None: ...from typing import Union, List, Optional, Callable, Iterator, TypedDict
from pathlib import Path
from port_for import PortType
import psycopg
from psycopg import Connection
from pytest import FixtureRequest, TempPathFactory
# Exception types
class ExecutableMissingException(FileNotFoundError): ...
class PostgreSQLUnsupported(Exception): ...