Asyncpg Recorder

Installation

uv add asyncpg-recorder

Usage

import asyncpg
from asyncpg_recorder import use_cassette


async def query():
    con = await asyncpg.connect(DSN)
    res = await con.fetch("SELECT NOW();")
    await con.close()
    return res


@use_cassette
def test_select_now_replay():
    query()

When using pytest parametrized fixtures put the @use_cassette decorator on the test function not the fixture:

import asyncpg
from asyncpg_recorder import use_cassette
import pytest
import pytest_asyncio


@pytest_asyncio.fixture(params=[False, True])
async def param(request):
    return request.param


@pytest.mark.asyncio
@pytest.mark.usefixtures("param")
@use_cassette
async def test_parametrized_fixtures(path):
    async def query():
      con = await asyncpg.connect(DSN)
      res = await con.fetch("SELECT NOW();")
      await con.close()
      return res

    await select_version()

If you want to save the cassettes in a specific directory, set the variable cassettes-dir in your pyproject.toml. The path should be relative to pyproject.toml.

[tool.asyncpg-recorder]
"cassettes-dir"="tests/cassettes"

Caveats

The same database request on Python >= 3.14 yields a different hash for cassette entry than on Python <= 3.13. If it is important to run tests against both Python version ranges you need to record twice.

Development

uv run prek install  # pre-commit
uv run pytest

Release

This project uses SemVer.

To make a new release run ./scripts/release.sh <version number>.

Limitation

• Works only with pytest
• Depends on testcontainers
  • Testcontainers is used to boot up a temporary Postgres instance to which asyncpg will be connected.
  • This slows test suite down.