Little wrapper around asyncpg for specific experience.

Basic usage

from asyncpg_engine import Engine

engine = await Engine.create("postgres://guest:[email protected]: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"

Development and contribution

First of all you should install Poetry using official instructions or solutions provided by your distro. Then install dependencies:

poetry install

Run PostgreSQL using provided docker-compose configuration:

docker-compose up  # run it in another terminal or add `-d` to daemonize

Project uses combination of flake8, black, isort and mypy for linting and pytest for testing.

poetry run flake8
poetry run mypy ./
poetry run pytest
GitHub - sivakov512/asyncpg-engine at
Little wrapper around asyncpg for specific experience. - GitHub - sivakov512/asyncpg-engine at