aiodal

---

An opinionated helper library for working with async sqlalchemy core, fastapi and auth0.

This library uses sqlalchemy>=2.

dal

The dal module is a wrapper around a sqlalchemy engine with some useful functionality for managing transactions. It is based loosely on patterns from Myers Essential Sqlalchemy which date back to before the ORM was introduced.

We prefer to use pydantic as our object mapping layer which is more flexible in development. The Sqlalchemy tables are reflected and held by a DataAccessLayer instance at object start.

engine = create_async_engine("...")
meta = sa.Metadata()

db = DataAccessLayer()
await db.reflect(engine, meta)

The instance is essentially just a proxy into the engine with a simpler API and easier access to some of the underlying objects.

Similarly we use a TransactionManager to proxy into an AsyncConnection instance.

async with db.engine.connect() as conn:
    transaction = TransactionManager(conn, db)

    #... do stuff with the transaction instance

Usually we use TransactionManager in an asynccontextmanager. We provide a simple one by default, though it is very easy to build this in an application. A similar pattern is common as a FastAPI dependency.

async with dal.transaction(db) as transaction:
    # ... do stuff

On both DataAccessLayer and TransactionManager we offer a few helper methods to get at some underlying object metadata such as unique constraints and tables. We reflect views by default and cache any TableValuedAlias and unique constraints for easier lookup.

The aiodal.connect module is provided to wrap some of this boilerplate for situations where you want to get set up quickly.

A json serializer and total count function are provided in aiodal.helpers. Total count is specifically used alongside aiodal.web.paginator to create well formated paginated responses using offset/limit. Examples of its use can be found in the tests.

pytest_plugin

Probably most useful is the pytest plugin. We've chosen to decouple it from alembic which means your dev environment will have to manage set up and tear down for any tests. You can look at our docker-compose.yaml file to see how this can be done using psql for isolating local tests.

There are a number of python fixutres provided to setup an engine, but the minimal conftest should just need the following.

# conftest.py

#1. Set anyio backend fixture scoped to session
@pytest.fixture(scope="session")
def anyio_backend():
    """Override this to use a different backend for testing"""
    return "asyncio"

# 2. provide the testdb you are connecting to.
@pytest.fixture(scope="session")
def engine_uri() -> str:
    return "postgresql+asyncpg://user:pass@host:port/dbname

To see the other overrides you can look at aiodal.pytest_plugin.

This give you access to an active transaction fixture that you can pass around. This is a common pattern found in alot of projects and is nothing new. Its default is to rollback after each test case.

web

The aiodal.web package has a few modules that we use specifically for integrating with fastapi and auth0.

• auth - dependency injections for working with auth0. It is a port of https://github.com/dorinclisu/fastapi-auth0 that we customized a bit.
• controllers - generic interfaces for querying data. Can be applied to standard CRUD ops. The Controller classes form an opionated set of query wrappers. In practice you'll probably want to make your own layer but these are useful for getting an app spun up quickly.
• models - various pydantic models to interact with forms, query parameters and provide base classes for different response objects.
• paginator - An offset limit pagination implementation that works with any return Result type from sqlalchemy
• slack_notify - Some tooling to wrap slack web hook calls in a reasonable matter. Can be used as a proxy or as middleware.
• version.py - Functionality to handle object versioning via Etag. Useful for optimistic and pessimistic setup.

A working application that shows these features can be found in tests.books and test.authapp.

usage

For now clone a tag from github. We consider this project "perma-Beta". The package evolves with our team's objectives. Expect breaking changes every minor release. These will be documented as best as possible.

Since v0.11 we are database agnostic though our testing effort and focus is on postgres via asyncpg.