Skip to main content

Wrapper around asyncpg with a bit better experience.

Project description


Small wrapper around asyncpg for specific experience and transactional testing.

test Status Coverage Status Code style: black Python versions PyPi

Basic usage

from asyncpg_engine import Engine

engine = await Engine.create("postgres://guest:guest@localhost:5432/guest?sslmode=disable")

async with engine.acquire() as con:
    assert await con.fetchval("SELECT 1") == 1

Custom type conversions

You can specify custom encoder/decoder by subclassing Engine:

from asyncpg_engine import Engine
import orjson

class MyEngine(Engine):

    async def _set_codecs(con: Connection) -> None:
        await con.set_type_codec(
            "json", encoder=orjson.dumps, decoder=orjson.loads, schema="pg_catalog"

Pytest plugin

Library includes pytest plugin with support for transactional testing.

To start using it install pytest, enable plugin in your root and define postgres_url fixture that returns database connection string:

pytest_plugins = ["asyncpg_engine"]

def postgres_url() -> str:
    return "postgres://guest:guest@localhost:5432/guest?sslmode=disable"

Now you can use two fixtures:

  • db that returns Engine instance:
async def test_returns_true(db):
    async with db.acquire() as con:
        assert await con.fetchval("SELECT true")
  • con that returns already acquired connection:
async def test_returns_true(con):
    assert await con.fetchval("SELECT true")

By default Engine is configured for transactional testing, so every call to db.acquire or con usage will return the same connection with already started transaction. Transaction is rolled back at the end of test, so all your changes in db are rolled back too.

You can override this behaviour with asyncpg_engine mark:

async def test_returns_true(con):
    assert await con.fetchval("SELECT true")

async def test_returns_true_too(db):
    async with db.acquire() as con:
        assert await con.fetchval("SELECT true")

If you want to use your own custom Engine subclass in tests you can define asyncpg_engine_cls fixture that returns it:

from asyncpg_engine import Engine

class MyPrettyEngine(Engine):

def asyncpg_engine_cls() -> typing.Type[MyPrettyEngine]:
    return MyPrettyEngine

async def test_returns_my_pretty_engine(db: MyPrettyEngine) -> None:
    assert isinstance(db, MyPrettyEngine)

Development and contribution

First of all you should install Poetry.

  • install project dependencies
make install
  • run linters
make lint
  • run tests
make test
  • feel free to contribute!

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

asyncpg_engine-0.3.2.tar.gz (5.0 kB view hashes)

Uploaded source

Built Distribution

asyncpg_engine-0.3.2-py3-none-any.whl (5.2 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page