FastAPI

# pip install fastapi "strawberry-graphql[fastapi]" "sqlalchemy[asyncio]" aiosqlite httpx2
# In production the URL is PostgreSQL, e.g. "postgresql+asyncpg://app@db/shop".
from contextlib import asynccontextmanager

import strawberry
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from strawberry.fastapi import GraphQLRouter

engine = create_async_engine("sqlite+aiosqlite:///:memory:")
Session = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

class OrderRow(Base):  # SQLAlchemy: a table
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    customer: Mapped[str]
    total_cents: Mapped[int]

@strawberry.type
class Order:  # Strawberry: a GraphQL type
    id: int
    total_cents: int

@strawberry.type
class Query:
    @strawberry.field
    async def orders(self, info: strawberry.Info, customer: str) -> list[Order]:
        session = info.context["session"]
        rows = await session.scalars(select(OrderRow).where(OrderRow.customer == customer))
        return [Order(id=r.id, total_cents=r.total_cents) for r in rows]

async def get_context():  # a FastAPI dependency: one session per request
    async with Session() as session:
        yield {"session": session}

@asynccontextmanager
async def lifespan(app: FastAPI):  # create and fill the table at startup
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with Session() as s:
        s.add_all([OrderRow(customer="ada", total_cents=1299),
                   OrderRow(customer="bob", total_cents=500)])
        await s.commit()
    yield

app = FastAPI(lifespan=lifespan)
app.include_router(GraphQLRouter(strawberry.Schema(Query), context_getter=get_context),
                   prefix="/graphql")

@app.get("/health")  # ordinary REST routes sit beside it
def health():
    return {"ok": True}

with TestClient(app) as client:
    query = '{ orders(customer: "ada") { id totalCents } }'
    print(client.post("/graphql", json={"query": query}).json())
    # {'data': {'orders': [{'id': 1, 'totalCents': 1299}]}}
    bad = client.post("/graphql", json={"query": "{ orders { id } }"})
    print(bad.status_code, bad.json()["errors"][0]["message"])
    # 200 Argument 'Query.orders(customer:)' of type 'String!' is required, but it was not provided.
    print(client.get("/health").json())  # {'ok': True}