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

这是什么？

这是一个 pytest 插件，允许您测试依赖于运行中的 PostgreSQL 数据库的代码。它提供了用于管理 PostgreSQL 进程和客户端连接的 fixtures。

快速入门

安装插件：

.. code-block:: sh

pip install pytest-postgresql

您还需要安装 psycopg (version 3). See its installation instructions <https://www.psycopg.org/psycopg3/docs/basic/install.html>_。

对于使用 psycopg.AsyncConnection 的异步测试，请安装可选的 async 扩展：

.. code-block:: sh

pip install pytest-postgresql[async]

这将安装：

pytest-asyncio (>= 1.4) — required for @pytest.mark.asyncio 和 postgresql_async 夹具。
aiofiles (>= 23.0) — 仅当通过异步加载器 (sql_async) 加载 SQL 文件时才需要。

在 Windows 上，当没有注册更早的 pytest-asyncio 循环工厂时，该插件会为 asyncio 测试自动配置一个 SelectorEventLoop。这是必需的，因为 psycopg 异步与默认的 ProactorEventLoop on Windows (documented by psycopg <https://www.psycopg.org/psycopg3/docs/advanced/async.html>_) 不兼容。没有它，postgresql_async tests fail with Psycopg 无法使用 'ProactorEventLoop' 以异步模式运行。安装 pytest-postgresql[async] 时无需额外配置。

在 Windows 上使用 pytest-asyncio >= 1.4 时，如果未提供先前的工厂，该插件会通过 pytest-asyncio 的 loop-factory 钩子为所有 asyncio 测试注册一个 selector 循环工厂。在 Python 3.14+ 上，由于该 API 已弃用，因此不使用旧的 asyncio 策略回退。

如果更早的钩子实现已经提供了循环工厂，则这些工厂将保持不变。使用先前工厂的测试可能在 pytest ID 中显示不同的循环名称（例如 test_example[custom] 而不是 test_example[selector]）。

如果在 Windows 上使用较旧的 pytest-asyncio (< 1.4) 且 Python < 3.14，该插件将回退到为整个测试会话设置全局 WindowsSelectorEventLoopPolicy — 而不仅仅是针对 postgresql 异步测试。这可能会改变同一次运行中不相关的 asyncio 测试的事件循环行为。安装 pytest-postgresql[async] (which pulls pytest-asyncio >= 1.4) 以避免该旧路径。

pytest-asyncio 配置

pytest-asyncio 1.x 默认值为 asyncio_mode = strict，因此每个异步测试都必须在 pytest.ini or 中标记为@pytest.mark.asyncio. If you set asyncio_mode = auto，未标记的异步测试函数会被自动检测 — 无需 postgresql_asyncstill requires the[async]`` 额外配置。

.. code-block:: ini

[pytest] asyncio_mode = strict

.. note::

虽然此插件需要 psycopg 3 to manage the database, your application code can still use `` 2。

运行测试：

只需包含 postgresql fixture in your test. It provides a connected psycopg.Connection 对象。

.. code-block:: python

def test_example(postgresql): """检查主要的 postgresql 夹具。""" with postgresql.cursor() as cur: cur.execute("CREATE TABLE test (id serial PRIMARY KEY, num integer, data varchar);") postgresql.commit()

对于异步代码，请使用 postgresql_async with pytest.mark.asyncio：

.. code-block:: python

import pytest

   @pytest.mark.asyncio
   async def test_example_async(postgresql_async):
       """检查主要的异步 postgresql fixture。"""
       async with postgresql_async.cursor() as cur:
           await cur.execute(
               "CREATE TABLE test (id serial PRIMARY KEY, num integer, data varchar);"
           )
           await postgresql_async.commit()

如何使用

.. warning::

已在 PostgreSQL 版本 >= 14 上测试。更多详情请参见测试。

工作原理

.. image:: https://raw.githubusercontent.com/dbfixtures/pytest-postgresql/main/docs/images/architecture.svg :alt: 项目架构图（同步 fixtures） :align: center

.. image:: https://raw.githubusercontent.com/dbfixtures/pytest-postgresql/main/docs/images/architecture_async.svg :alt: 项目架构图（异步 fixtures） :align: center

该插件提供两种主要类型的 fixtures：

1. 客户端 Fixtures 这些为测试提供数据库连接。

* **postgresql** - 一个函数作用域的 fixture。它返回一个已连接的 ``psycopg.Connection``。
  在每个测试之后，它会终止剩余的连接并删除测试数据库，以确保隔离性。
* **postgresql_async** - 异步对应版本。它返回一个已连接的 ``psycopg.AsyncConnection``。
  需要 ``pytest-postgresql[async]`` (``pytest-asyncio`` >= 1.4)，并且每个测试都必须
  标记为 ``@pytest.mark.asyncio``。

异步 Fixtures postgresql_async and custom factories created with factories.postgresql_async 是使用 pytest_asyncio.fixture 的异步生成器 fixtures。

如果手动安装而不是通过 ``[async]`` 安装，最低版本为：

.. code-block:: text

pytest-asyncio >= 1.4 aiofiles >= 23.0 # 仅用于异步 SQL 文件加载

如果 ``pytest-asyncio`` is missing, fixture setup raises ``ImportError``。

**异步 SQL 文件加载**

Process 和 noproc fixtures 始终在会话设置期间同步填充其模板数据库
（通过 ``DatabaseJanitor.load()``），即使你在
process fixture 的 ``load`` list are executed with the sync ``sql()`` 加载器中使用了 ``postgresql_async`` as the client fixture.  SQL ``Path`` 条目。

当你
调用 ``AsyncDatabaseJanitor.load()`` directly with a ``Path`` 时，请使用 ``sql_async`` (requires ``aiofiles`` from the ``[async]`` 额外参数。
传递给 ``AsyncDatabaseJanitor.load()`` 的可调用加载器可以是同步或异步的；
返回的可等待值会被自动等待。

.. code-block:: python

from pathlib import Path from pytest_postgresql import factories

    postgresql_my_proc = factories.postgresql_proc(load=[Path("schema.sql")])
    postgresql_my_async = factories.postgresql_async("postgresql_my_proc")

2. Process Fixtures 这些管理 PostgreSQL 服务器的生命周期。

* **postgresql_proc** - 一个会话范围的 fixture，在首次使用时启动一个 PostgreSQL 实例，并在所有测试完成后停止它。
* **postgresql_noproc** - 一个用于连接已运行的 PostgreSQL 实例的 fixture（例如，在 Docker 或 CI 中）。

Customizing Fixtures

你可以使用 factories 创建额外的 fixtures：

.. code-block:: python

from pytest_postgresql import factories

# 创建自定义进程 fixture
postgresql_my_proc = factories.postgresql_proc(
    port=None, unixsocketdir='/var/run')

# 创建使用自定义进程的客户端 fixture
postgresql_my = factories.postgresql('postgresql_my_proc')

# 异步客户端 fixture（需要 pytest-postgresql[async]、pytest-asyncio >= 1.4）
postgresql_my_async = factories.postgresql_async('postgresql_my_proc')

.. note::

每个进程 fixture 都可以通过工厂参数独立配置。

为测试预填充数据库

如果你希望数据库自动预填充你的 schema 和数据，可以在两个层级实现：

#. 每个测试： 在客户端 fixture 中，通过使用中间 fixture。 #. 每个会话： 在进程 fixture 中。

进程 fixture 接受一个 load 参数，该参数支持：

SQL 文件路径： 加载并执行 SQL 文件。
加载函数： 一个可调用对象或导入字符串（例如，"path.to.module:function"）。这些函数接收 host、port、user、dbname 和 password，并且必须自行执行连接（或使用 ORM）。

进程 fixture 每个会话将数据库预填充一次到一个模板数据库中。然后客户端 fixture 为每个测试克隆此模板，从而显著加快你的测试速度。

.. code-block:: python

from pathlib import Path postgresql_my_proc = factories.postgresql_proc( load=[ Path("schemafile.sql"), "import.path.to.function", load_this_callable ] )

在命令行上定义预填充：

.. code-block:: sh

pytest --postgresql-load=path/to/file.sql --postgresql-load=path.to.function

如果加载的 .sql 文件包含无法在事务块中运行的语句（例如 CREATE DATABASE），请在加载器连接上启用自动提交。这可以通过工厂参数、命令行或 pytest.ini 进行设置，而无需编写自定义加载器：

.. code-block:: python

postgresql_my_proc = factories.postgresql_proc(load=[Path("with_create_db.sql")], load_autocommit=True)

.. code-block:: sh

pytest --postgresql-load-autocommit

连接到现有的 PostgreSQL 数据库

要连接到外部服务器（例如，在 Docker 中运行的服务器），请使用 postgresql_noproc fixture。

对于针对外部服务器的异步测试，请使用 factories.postgresql_async("postgresql_noproc") 创建客户端 fixture（参见 tests/examples/test_drop_test_database_async.py）。

.. code-block:: python

postgresql_external = factories.postgresql('postgresql_noproc')

默认情况下，它连接到 127.0.0.1:5432。

使用 `postgres` 以外的维护数据库

创建和删除测试数据库需要一个已存在的数据库的连接 - 默认情况下是 postgres。如果测试角色仅缺少权限， GRANT CONNECT ON DATABASE postgres TO myuser is the simpler fix. When `` 完全不可达 - 例如在具有固定数据库列表的连接池器之后，或在未公开它的托管服务器上 - 请将 noproc fixture 指向另一个数据库：

.. code-block:: sh

pytest --postgresql-maintenance-dbname=my_existing_db

.. code-block:: python

postgresql_external = factories.postgresql_noproc(maintenance_dbname="my_existing_db")

数据库仅会被连接，永远不会被创建、修改或删除。避免使用 template1：当未提供模板时，CREATE DATABASE 会对其进行克隆，而 PostgreSQL 不期望源数据库在复制过程中处于使用状态。

Chaining fixtures

你可以链式使用多个 postgresql_noproc fixtures 来分层预填充数据。链中的每个 fixture 都会基于前一个 fixture 创建自己的模板数据库。

.. code-block:: python

from pytest_postgresql import factories

# 1. 从一个进程或无进程基础开始
base_proc = factories.postgresql_proc(load=[load_schema])

# 2. 添加一个包含一些数据的层
seeded_noproc = factories.postgresql_noproc(depends_on="base_proc", load=[load_data])

# 3. 添加另一个包含更多数据的层
more_seeded_noproc = factories.postgresql_noproc(depends_on="seeded_noproc", load=[load_more_data])

# 4. 在测试中使用最终层
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

您可以通过 fixture 工厂参数、命令行选项或 pytest.ini 来定义设置。它们按以下顺序解析：

Fixture factory argument
Command line option
pytest.ini configuration option

.. list-table:: 配置选项 :header-rows: 1

- 设置
  - postgresql_proc 参数
  - postgresql_noproc 参数
  - 命令行选项
  - pytest.ini 选项
  - 默认值
- - 可执行文件路径
  - executable
  - n/a
  - --postgresql-exec
  - postgresql_exec
  - pg_config --bindir + pg_ctl
- - host
  - host
  - host
  - --postgresql-host
  - postgresql_host
  - 127.0.0.1
- - port
  - port
  - port
  - --postgresql-port
  - postgresql_port
  - random (proc), 5432 (noproc)
- - 端口搜索次数
  - —
  - n/a
  - --postgresql-port-search-count
  - postgresql_port_search_count
  - 5
- - postgresql 用户
  - user
  - user
  - --postgresql-user
  - postgresql_user
  - postgres
- - 密码
  - password
  - password
  - --postgresql-password
  - postgresql_password
- - 启动参数（额外的 pg_ctl 参数）
  - startparams
  - n/a
  - --postgresql-startparams
  - postgresql_startparams
  - -w
- - Postgres 可执行文件额外参数（通过 pg_ctl 的 -o 参数传递）
  - postgres_options
  - n/a
  - --postgresql-postgres-options
  - postgresql_postgres_options
- - unixsocket 位置
  - unixsocket
  - n/a
  - --postgresql-unixsocketdir
  - postgresql_unixsocketdir
  - $TMPDIR
- - 数据库名称
  - dbname
  - dbname (handles xdist)
  - --postgresql-dbname
  - postgresql_dbname
  - tests
- - 维护数据库名称
  - n/a

maintenance_dbname
- --postgresql-maintenance-dbname
- postgresql_maintenance_dbname
- postgres
- - 默认模式（加载列表）
  - load
  - load
  - --postgresql-load
  - postgresql_load
- - SQL 加载器连接的自动提交
  - load_autocommit
  - load_autocommit
  - --postgresql-load-autocommit
  - postgresql_load_autocommit
  - False
- - PostgreSQL 连接选项
  - options
  - options
  - --postgresql-options
  - postgresql_options
- - 启动时删除测试数据库
  - —
  - —
  - --postgresql-drop-test-database
  - false
- - 模板数据库所依赖的 Fixture
  - n/a
  - depends_on

— 表示该设置适用于该 fixture，但没有工厂参数； n/a 表示该设置完全不适用于该 fixture。

.. note::

如果 executable is not provided, the plugin attempts to find it by calling pg_config. If that fails, it falls back to a common path like /usr/lib/postgresql/14/bin/pg_ctl。

示例

使用 SQLAlchemy

本示例展示了如何创建一个 SQLAlchemy 会话 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]:
    """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)

    # 假设你使用了 Base 模型
    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)

高级用法：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",
    ):
        with psycopg.connect(
            dbname="my_custom_db",
            user=postgresql_proc.user,
            host=postgresql_proc.host,
            port=postgresql_proc.port,
            password="secret_password",
        ) as conn:
            # use connection
            pass

高级用法：AsyncDatabaseJanitor

AsyncDatabaseJanitor is the async counterpart to DatabaseJanitor。在标准 fixture 之外使用 psycopg.AsyncConnection 管理数据库状态时使用它。它需要 psycopg（一个核心依赖项）。安装 pytest-postgresql[async] when you need aiofiles 以通过 sql_async, or pytest-asyncio 加载 SQL 文件，用于 pytest 异步测试。

.. code-block:: python

import pytest import psycopg from pytest_postgresql.janitor import AsyncDatabaseJanitor

@pytest.mark.asyncio
async def test_manual_async_janitor(postgresql_proc):
    async with AsyncDatabaseJanitor(
        user=postgresql_proc.user,
        host=postgresql_proc.host,
        port=postgresql_proc.port,
        dbname="my_custom_db",
        version=postgresql_proc.version,
        password="secret_password",
    ):
        async with await psycopg.AsyncConnection.connect(
            dbname="my_custom_db",
            user=postgresql_proc.user,
            host=postgresql_proc.host,
            port=postgresql_proc.port,
            password="secret_password",
        ) as conn:
            # 使用异步连接
            pass

连接到 Docker 中的 PostgreSQL

要连接到通过 Docker 运行的 PostgreSQL，请使用 noproc fixture。

.. code-block:: sh

docker run --name some-postgres -e POSTGRES_PASSWORD=mysecret -d postgres

在你的测试中：

.. code-block:: python

from pytest_postgresql import factories

postgresql_in_docker = factories.postgresql_noproc()
postgresql = factories.postgresql("postgresql_in_docker", dbname="test")

def test_docker(postgresql):
    with postgresql.cursor() as cur:
        cur.execute("SELECT 1")

运行方式：

.. code-block:: sh

pytest --postgresql-host=172.17.0.2 --postgresql-password=mysecret

所有测试的基础数据库状态

您可以定义一个 load 函数并将其传递给您的进程 fixture 工厂：

.. code-block:: python

import psycopg from pytest_postgresql import factories

def load_database(**kwargs):
    with psycopg.connect(**kwargs) as conn:
        with conn.cursor() as cur:
            cur.execute("CREATE TABLE stories (id serial PRIMARY KEY, name varchar);")
            cur.execute("INSERT INTO stories (name) VALUES ('Silmarillion'), ('The Expanse');")

postgresql_proc = factories.postgresql_proc(load=[load_database])
postgresql = factories.postgresql("postgresql_proc")

def test_stories(postgresql):
    with postgresql.cursor() as cur:
        cur.execute("SELECT count(*) FROM stories")
        assert cur.fetchone()[0] == 2

进程 fixture 会填充一次 模板数据库，而客户端 fixture 会为每个测试克隆它。这既快速又干净，并确保没有悬挂事务。这种方法适用于 postgresql_proc and postgresql_noproc。

pytest-postgresql

这是什么？

快速入门

如何使用

工作原理

Customizing Fixtures

为测试预填充数据库

连接到现有的 PostgreSQL 数据库

使用 postgres 以外的维护数据库

Chaining fixtures

Configuration

示例

使用 SQLAlchemy

高级用法：DatabaseJanitor

高级用法：AsyncDatabaseJanitor

连接到 Docker 中的 PostgreSQL

所有测试的基础数据库状态

使用 `postgres` 以外的维护数据库