Wrapper around asyncpg with a bit better experience.
Project description
asyncpg-engine
Little wrapper around asyncpg for specific experience.
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:
# 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 encoder\decoder 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 testing.
poetry run flake8
poetry run mypy ./
poetry run pytest
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.1.0.tar.gz
(3.9 kB
view hashes)
Built Distribution
Close
Hashes for asyncpg_engine-0.1.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | d9ac64fa65e58d62670fa88e1e83b5bb2b0b5dccf10385b726f0a3a60e986bda |
|
MD5 | 712d0d4bad73681036937a091526854a |
|
BLAKE2b-256 | 225b990d9569093de18219e09a50611c2b71b1a04aee4806dc4e3b5e90735ce6 |