Wrapper around asyncpg with a bit better experience.
Project description
asyncpg-engine
Little wrapper around asyncpg for specific experience and transactional testing.
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"
)
Pytest plugin
Library includes pytest plugin with support for transactional testing.
To start using it install pytest
, enable plugin in your root conftest.py
and define postgres_url
fixture that returns connection string to database:
pytest_plugins = ["asyncpg_engine"]
@pytest.fixture()
def postgres_url() -> str:
return "postgres://guest:guest@localhost:5432/guest?sslmode=disable"
Now you can use two fixtures:
db
that returnsEngine
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
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:
@pytest.mark.asyncpg_engine(transactional=False)
async def test_returns_true(con):
assert await con.fetchval("SELECT true")
@pytest.mark.asyncpg_engine(transactional=False)
async def test_returns_true_too(db):
async with db.acquire() as con:
assert await con.fetchval("SELECT true")
If you want to use in tests your own custom Engine
subclass you can define asyncpg_engine_cls
fixture that returns it:
from asyncpg_engine import Engine
class MyPrettyEngine(Engine):
pass
@pytest.fixture()
def asyncpg_engine_cls() -> t.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 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
Built Distribution
Hashes for asyncpg_engine-0.2.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 063d8e0359a9a873eb30ecfd03a3ac4c32c29f30a7cc0c17027a4c91eadd3846 |
|
MD5 | d1c0b191b8316062a3a596244101dd23 |
|
BLAKE2b-256 | 3e7db3e34adedc6899e11b48e989206d19f17767bb22aac1bfbc8ee6a731b834 |