Skip to content

Instantly share code, notes, and snippets.

@pydanny

pydanny/air_asyncpg.py

Last active November 26, 2025 05:09
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save pydanny/e2be1d4c77c8365f53b8fe2798c8d9ad to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save pydanny/e2be1d4c77c8365f53b8fe2798c8d9ad to your computer and use it in GitHub Desktop.
Download ZIP
Raw
air_asyncpg.py
from typing import AsyncIterator
import asyncpg
from ..applications import Air
from contextlib import asynccontextmanager
from typing import Callable
from os import getenv
DATABASE_URL = getenv('DATABASE_URL')
def make_lifespan(dsn: str, *, min_size: int = 1, max_size: int = 10) -> Callable:
@asynccontextmanager
async def lifespan(app: Air) -> AsyncIterator[None]:
app.state.pool = await asyncpg.create_pool(
dsn=dsn,
min_size=min_size,
max_size=max_size,
)
try:
yield
finally:
await app.state.pool.close()
return lifespan
@pydanny
Copy link
Author

pydanny commented Nov 26, 2025 •
edited
Loading

A better, more robust version is:

import asyncpg
from contextlib import asynccontextmanager
from typing import Callable
from os import getenv

import air

DATABASE_URL = getenv('DATABASE_URL')


def make_lifespan(dsn: str, *, min_size: int = 1, max_size: int = 10) -> Callable:
    @asynccontextmanager
    async def lifespan(app: Air):
        async with asyncpg.create_pool(
            dsn=dsn,
            min_size=min_size,
            max_size=max_size,
        ) as pool:
            yield {'pool': pool}

    return lifespan

lifespan = make_lifespan(DATABASE_URL)
app = air.Air(lifespan=lifespan)

Usage:

@app.page
async def index(request: Request):
    async with request.state.pool.acquire() as conn:
        users = await conn.fetchmany(
            "SELECT * FROM users;"
        )
    return *[air.P(u) for x in users]

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment