An example how to combine async fastapi server with async psycopg

fastapi-posgresql.py

#!/usr/bin/env python3

# to run use:
#     $ fastapi dev fastapi-posgresql.py
from contextlib import asynccontextmanager
from typing import Annotated, AsyncGenerator

import psycopg
from psycopg_pool import AsyncConnectionPool

from fastapi import Depends, FastAPI

# create a shared object of connection pools to the db
# from which every request would grab a connection
pool = AsyncConnectionPool("dbname=example", open=False)


# every request would run this function to get DB connection
# from the above pool
async def get_db() -> AsyncGenerator[psycopg.AsyncConnection, None]:
    async with pool.connection() as conn:
        # the coneection would be yielded here
        yield conn
        # then after request is done get_db would continue from here
        # closing the connection


# lifespan would initialize and destroy per-server context
@asynccontextmanager
async def lifespan(_: FastAPI):
    # this runs when server is loaded, and keep running untiil yield
    await pool.open()
    yield
    # this runs when server shuts down
    await pool.close()


# here we tell fastapi to use the generator above to initialize server
app = FastAPI(lifespan=lifespan)


# The argument is annotated with `Depends`, which means the generator
# above, `get_db`, would run for this type of request and fetch a connection
# from the pool
@app.get("/")
async def root(conn: Annotated[psycopg.AsyncConnection, Depends(get_db)]):
    async with conn.cursor() as cur:
        await cur.execute("select * from information_schema.tables")
        return await cur.fetchall()