Little wrapper around asyncpg for specific experience
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:
# https://magicstack.github.io/asyncpg/current/api/index.html#asyncpg.connection.Connection
assert await con.fetchval("SELECT 1") == 1
Custom type conversions
You can specify custom encoderdecoder by subclassing Engine:
from asyncpg_engine import Engine
import orjson
class MyEngine(Engine):
@staticmethod
async def _set_codecs(con: Connection) -> None:
# https://magicstack.github.io/asyncpg/current/api/index.html#asyncpg.connection.Connection.set_type_codec
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