Getting started

Configure the connection to the database

for example for PostgreSQL database.py:

from sqlalchemy.ext.asyncio import (
    async_sessionmaker,
    AsyncEngine,
    AsyncSession,
    create_async_engine,
)

from context_async_sqlalchemy import DBConnect


def create_engine(host: str) -> AsyncEngine:
    """
    database connection parameters.
    """

    # In production code, you will probably take these parameters from env
    pg_user = "krylosov-aa"
    pg_password = ""
    pg_port = 6432
    pg_db = "test"
    return create_async_engine(
        f"postgresql+asyncpg://"
        f"{pg_user}:{pg_password}"
        f"@{host}:{pg_port}"
        f"/{pg_db}",
        future=True,
        pool_pre_ping=True,
    )


def create_session_maker(
    engine: AsyncEngine,
) -> async_sessionmaker[AsyncSession]:
    """session parameters"""
    return async_sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False
    )


connection = DBConnect(
    host="127.0.0.1",
    engine_creator=create_engine,
    session_maker_creator=create_session_maker,
)

Manage Database connection lifecycle

Close the resources at the end of your application's life

Example for FastAPI:

from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator
from fastapi import FastAPI

from database import connection


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, Any]:
    """Database connection lifecycle management"""
    yield
    await connection.close()  # Close the engine if it was open

Setup middleware

Middleware takes on the most important and complex work of managing context and sessions.

You can use ready-made middlewares:

FastAPI

from context_async_sqlalchemy.fastapi_utils import (
    add_fastapi_http_db_session_middleware,
)

app = FastAPI(...)
add_fastapi_http_db_session_middleware(app)

Starlette

from context_async_sqlalchemy.starlette_utils import (
    add_starlette_http_db_session_middleware,
)

app = Starlette(...)
add_starlette_http_db_session_middleware(app)

Write own

If there is no ready-made solution for you, don't worry! You can see how it works and write your own.

Use it

from context_async_sqlalchemy import db_session
from sqlalchemy import insert

from database import connection  # your configured connection to the database
from models import ExampleTable  # just some model for example

async def some_func() -> None:
    # Created a session (no connection to the database yet)
    session = await db_session(connection)

    stmt = insert(ExampleTable).values(text="example_with_db_session")

    # On the first request, a connection and transaction were opened
    await session.execute(stmt)

    # If you call db_session again, it will return the same session
    # even in child coroutines.
    session = await db_session(connection)

    # The second request will use the same connection and the same transaction
    await session.execute(stmt)

    # The commit and closing of the session will occur automatically

Examples

The repository includes an example integration with FastAPI, which describes numerous workflows. FastAPI example

It also includes two types of test setups you can use in your projects.

All library tests are in the examples, as we want to test not in the abstract but in the context of a real asynchronous web application.

FastAPI tests example