pytest-postgresql
Postgresql fixtures and fixture factories for Pytest.
Rank: #1901Downloads: 4,224,136 (30 days)Stars: 514Forks: 53
Description
.. image:: https://raw.githubusercontent.com/dbfixtures/pytest-postgresql/main/logo.png
:width: 100px
:height: 100px
pytest-postgresql
=================
.. image:: https://img.shields.io/pypi/v/pytest-postgresql.svg
:target: https://pypi.python.org/pypi/pytest-postgresql/
:alt: Latest PyPI version
.. image:: https://img.shields.io/pypi/wheel/pytest-postgresql.svg
:target: https://pypi.python.org/pypi/pytest-postgresql/
:alt: Wheel Status
.. image:: https://img.shields.io/pypi/pyversions/pytest-postgresql.svg
:target: https://pypi.python.org/pypi/pytest-postgresql/
:alt: Supported Python Versions
.. image:: https://img.shields.io/pypi/l/pytest-postgresql.svg
:target: https://pypi.python.org/pypi/pytest-postgresql/
:alt: License
What is this?
=============
This is a pytest plugin that enables you to test code relying on a running PostgreSQL database.
It provides fixtures for managing both the PostgreSQL process and the client connections.
Quick Start
===========
1. **Install the plugin:**
.. code-block:: sh
pip install pytest-postgresql
You will also need to install ``psycopg`` (version 3). See `its installation instructions <https://www.psycopg.org/psycopg3/docs/basic/install.html>`_.
.. note::
While this plugin requires ``psycopg`` 3 to manage the database, your application code can still use ``psycopg`` 2.
2. **Run a test:**
Simply include the ``postgresql`` fixture in your test. It provides a connected ``psycopg.Connection`` object.
.. code-block:: python
def test_example(postgresql):
"""Check main postgresql fixture."""
with postgresql.cursor() as cur:
cur.execute("CREATE TABLE test (id serial PRIMARY KEY, num integer, data varchar);")
postgresql.commit()
How to use
==========
.. warning::
Tested on PostgreSQL versions >= 14. See tests for more details.
How does it work
----------------
.. image:: https://raw.githubusercontent.com/dbfixtures/pytest-postgresql/main/docs/images/architecture.svg
:alt: Project Architecture Diagram
:align: center
The plugin provides two main types of fixtures:
**1. Client Fixtures**
These provide a connection to a database for your tests.
* **postgresql** - A function-scoped fixture. It returns a connected ``psycopg.Connection``.
After each test, it terminates leftover connections and drops the test database to ensure isolation.
**2. Process Fixtures**
These manage the PostgreSQL server lifecycle.
* **postgresql_proc** - A session-scoped fixture that starts a PostgreSQL instance on its first use and stops it when all tests are finished.
* **postgresql_noproc** - A fixture for connecting to an already running PostgreSQL instance (e.g., in Docker or CI).
Customizing Fixtures
--------------------
You can create additional fixtures using factories:
.. code-block:: python
from pytest_postgresql import factories
# Create a custom process fixture
postgresql_my_proc = factories.postgresql_proc(
port=None, unixsocketdir='/var/run')
# Create a client fixture that uses the custom process
postgresql_my = factories.postgresql('postgresql_my_proc')
.. note::
Each process fixture can be configured independently through factory arguments.
Pre-populating the database for tests
-------------------------------------
If you want the database to be automatically pre-populated with your schema and data, there are two levels you can achieve it:
#. **Per test:** In a client fixture, by using an intermediary fixture.
#. **Per session:** In a process fixture.
The process fixture accepts a ``load`` parameter, which supports:
* **SQL file paths:** Loads and executes the SQL files.
* **Loading functions:** A callable or an import string (e.g., ``"path.to.module:function"``).
These functions receive **host**, **port**, **user**, **dbname**, and **password** and must perform the connection themselves (or use an ORM).
The process fixture pre-populates the database once per session into a **template database**. The client fixture then clones this template for each test, which significantly **speeds up your tests**.
.. code-block:: python
from pathlib import Path
postgresql_my_proc = factories.postgresql_proc(
load=[
Path("schemafile.sql"),
"import.path.to.function",
load_this_callable
]
)
Defining pre-population on the command line:
.. code-block:: sh
pytest --postgresql-populate-template=path/to/file.sql --postgresql-populate-template=path.to.function
Connecting to an existing PostgreSQL database
----------------------------------------------
To connect to an external server (e.g., running in Docker), use the ``postgresql_noproc`` fixture.
.. code-block:: python
postgresql_external = factories.postgresql('postgresql_noproc')
By default, it connects to ``127.0.0.1:5432``.
Chaining fixtures
-----------------
You can chain multiple ``postgresql_noproc`` fixtures to layer your data pre-population. Each fixture in the chain will create its own template database based on the previous one.
.. code-block:: python
from pytest_postgresql import factories
# 1. Start with a process or a no-process base
base_proc = factories.postgresql_proc(load=[load_schema])
# 2. Add a layer with some data
seeded_noproc = factories.postgresql_noproc(depends_on="base_proc", load=[load_data])
# 3. Add another layer with more data
more_seeded_noproc = factories.postgresql_noproc(depends_on="seeded_noproc", load=[load_more_data])
# 4. Use the final layer in your test
client = factories.postgresql("more_seeded_noproc")
.. image:: https://raw.githubusercontent.com/dbfixtures/pytest-postgresql/main/docs/images/architecture_chaining.svg
:alt: Fixture Chaining Diagram
:align: center
Configuration
=============
You can define settings via fixture factory arguments, command line options, or ``pytest.ini``. They are resolved in this order:
1. ``Fixture factory argument``
2. ``Command line option``
3. ``pytest.ini configuration option``
.. list-table:: Configuration options
:header-rows: 1
* - PostgreSQL option
- Fixture factory argument
- Command line option
- pytest.ini option
- Noop process fixture
- Default
* - Path to executable
- executable
- --postgresql-exec
- postgresql_exec
- -
- ``pg_config --bindir`` + ``pg_ctl``
* - host
- host
- --postgresql-host
- postgresql_host
- yes
- 127.0.0.1
* - port
- port
- --postgresql-port
- postgresql_port
- yes (5432)
- random
* - Port search count
-
- --postgresql-port-search-count
- postgresql_port_search_count
- -
- 5
* - postgresql user
- user
- --postgresql-user
- postgresql_user
- yes
- postgres
* - password
- password
- --postgresql-password
- postgresql_password
- yes
-
* - Starting parameters (extra pg_ctl arguments)
- startparams
- --postgresql-startparams
- postgresql_startparams
- -
- -w
* - Postgres exe extra arguments (passed via pg_ctl's -o argument)
- postgres_options
- --postgresql-postgres-options
- postgresql_postgres_options
- -
-
* - Location for unixsockets
- unixsocket
- --postgresql-unixsocketdir
- postgresql_unixsocketdir
- -
- $TMPDIR
* - Database name
- dbname
- --postgresql-dbname
- postgresql_dbname
- yes (handles xdist)
- test
* - Default Schema (load list)
- load
- --postgresql-load
- postgresql_load
- yes
-
* - PostgreSQL connection options
- options
- --postgresql-options
- postgresql_options
- yes
-
* - Drop test database on start
-
- --postgresql-drop-test-database
-
- -
- false
.. note::
If the ``executable`` is not provided, the plugin attempts to find it by calling ``pg_config``. If that fails, it fallbacks to a common path like ``/usr/lib/postgresql/13/bin/pg_ctl``.
Examples
========
Using SQLAlchemy
----------------
This example shows how to create an SQLAlchemy session fixture:
.. code-block:: python
from typing import Iterator
import pytest
from psycopg import Connection
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker, scoped_session
from sqlalchemy.pool import NullPool
@pytest.fixture
def db_session(postgresql: Connection) -> Iterator[Session]:
"""Session for SQLAlchemy."""
user = postgresql.info.user
host = postgresql.info.host
port = postgresql.info.port
dbname = postgresql.info.dbname
connection_str = f'postgresql+psycopg://{user}:@{host}:{port}/{dbname}'
engine = create_engine(connection_str, echo=False, poolclass=NullPool)
# Assuming you use a Base model
from my_app.models import Base
Base.metadata.create_all(engine)
SessionLocal = scoped_session(sessionmaker(bind=engine))
yield SessionLocal()
SessionLocal.close()
Base.metadata.drop_all(engine)
Advanced Usage: DatabaseJanitor
-------------------------------
``DatabaseJanitor`` is an advanced API for managing database state outside of standard fixtures. It is used by projects like `Warehouse <https://github.com/pypa/warehouse>`_ (pypi.org).
.. code-block:: python
import psycopg
from pytest_postgresql.janitor import DatabaseJanitor
def test_manual_janitor(postgresql_proc):
with DatabaseJanitor(
user=postgresql_proc.user,
host=postgresql_proc.host,
port=postgresql_proc.port,
dbname="my_custom_db",
version=postgresql_proc.version,
password="secret_password",