FastAPI

pip install "goatnotify[fastapi]"

Setup

Build the notifier once and attach it with the lifespan:

from fastapi import FastAPI

from goatnotify import AsyncioQueue, EmailChannel, Mailer, Notifier, SmtpTransport
from goatnotify.fastapi import NotifierDep, lifespan

notifier = Notifier(
    [EmailChannel(Mailer(SmtpTransport("smtp.example.com"), sender="noreply@example.com"))],
    queue=AsyncioQueue(),
)
app = FastAPI(lifespan=lifespan(notifier))


@app.post("/signup")
async def signup(payload: SignupForm, notifier: NotifierDep) -> dict[str, str]:
    user = await create_user(payload)
    await notifier.enqueue(user, Welcome())
    return {"status": "ok"}

NotifierDep is Annotated[Notifier, Depends(get_notifier)]. Using it without the lifespan raises ConfigurationError.

Background delivery

With an AsyncioQueue, enqueue() returns immediately and the notification is delivered in the background. On shutdown, the lifespan drains the queue: it waits for pending notifications and leaves the queue usable, so the same app can start again (as test clients do). Notifications still running after shutdown_timeout (30 seconds by default, None to wait forever) are cancelled and logged: lifespan(notifier, shutdown_timeout=10).

Composing with your own lifespan

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from goatnotify.fastapi import attach


@asynccontextmanager
async def app_lifespan(app: FastAPI) -> AsyncGenerator[None]:
    async with database.connect(), attach(app, notifier):
        yield


app = FastAPI(lifespan=app_lifespan)