GraphQL — FastAPI Pro: От новичка до архитектора

GraphQL с FastAPI

GraphQL — альтернатива REST для гибких запросов.

#REST vs GraphQL

Критерий	REST	GraphQL
Запросы	Фиксированные endpoints	Гибкий запрос
Over-fetching	Получаешь лишние данные	Только нужные поля
Under-fetching	Несколько запросов	Один запрос
Версионирование	/api/v1, /api/v2	Эволюция схемы

#Установка Strawberry

pip install strawberry-graphql fastapi

#Базовый пример

import strawberry
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

@strawberry.type
class User:
    id: int
    name: str
    email: str

@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: int) -> User:
        return User(id=id, name="John", email="john@example.com")
    
    @strawberry.field
    def users(self) -> list[User]:
        return [
            User(id=1, name="John", email="john@example.com"),
            User(id=2, name="Jane", email="jane@example.com")
        ]

schema = strawberry.Schema(query=Query)
app = FastAPI()
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

#Запрос

query {
  user(id: 1) {
    id
    name
    email
  }
  
  users {
    id
    name
  }
}

#Мутации (изменения)

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, name: str, email: str) -> User:
        # Создание пользователя
        return User(id=1, name=name, email=email)
    
    @strawberry.mutation
    def delete_user(self, id: int) -> bool:
        # Удаление
        return True

schema = strawberry.Schema(query=Query, mutation=Mutation)

#Вложенные типы

@strawberry.type
class Post:
    id: int
    title: str
    content: str

@strawberry.type
class User:
    id: int
    name: str
    email: str
    
    @strawberry.field
    def posts(self) -> list[Post]:
        return [
            Post(id=1, title="Post 1", content="Content 1"),
            Post(id=2, title="Post 2", content="Content 2")
        ]

# Запрос:
# query {
#   user(id: 1) {
#     name
#     posts {
#       title
#       content
#     }
#   }
# }

#Input типы

@strawberry.input
class CreateUserInput:
    name: str
    email: str
    age: int | None = None

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, input: CreateUserInput) -> User:
        return User(id=1, name=input.name, email=input.email)

#Error handling

@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: int) -> User | None:
        user = db.get_user(id)
        if not user:
            return None
        return user

# Или с ошибками
@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: int) -> User:
        user = db.get_user(id)
        if not user:
            raise Exception("User not found")
        return user

#DataLoader (N+1 проблема)

from strawberry.dataloader import DataLoader

async def batch_load_users(keys: list[int]) -> list[User]:
    users = db.query(User).filter(User.id.in_(keys)).all()
    return [next((u for u in users if u.id == k), None) for k in keys]

@strawberry.type
class Query:
    @strawberry.field
    async def get_users(self, ids: list[int]) -> list[User]:
        loader = DataLoader(batch_load_users)
        return await loader.load_many(ids)

#Аутентификация

from fastapi import Request

async def get_context(request: Request) -> dict:
    return {"user": request.state.user}

@strawberry.type
class Query:
    @strawberry.field
    def me(self, info: strawberry.Info) -> User | None:
        context = info.context
        if not context["user"]:
            return None
        return context["user"]

graphql_app = GraphQLRouter(schema, context_getter=get_context)

#Подписки (real-time)

import asyncio
import strawberry
from typing import AsyncGenerator

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def count(self, target: int = 10) -> AsyncGenerator[int, None]:
        for i in range(target):
            yield i
            await asyncio.sleep(1)

schema = strawberry.Schema(query=Query, subscription=Subscription)

#Интеграция с SQLAlchemy

from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@strawberry.type
class Query:
    @strawberry.field
    def users(self, info: strawberry.Info) -> list[User]:
        db: Session = info.context["db"]
        return db.query(UserModel).all()

async def get_context(db: Session = Depends(get_db)) -> dict:
    return {"db": db}

graphql_app = GraphQLRouter(schema, context_getter=get_context)

#Гибридный подход (REST + GraphQL)

app = FastAPI()

# REST endpoints
@app.get('/api/users')
def get_users_rest():
    return db.query(User).all()

# GraphQL
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

#Исполняемая лаборатория

Лаборатория «Strawberry GraphQL, Query/Mutation, over-fetching» доступна в публичном GitLab. Она проверяет Strawberry GraphQL с типами, Query и Mutation, вложенные типы с DataLoader для устранения N+1, аутентификацию через context и сравнение over-fetching с REST.

Склонируйте репозиторий и запустите тесты из его корня:

git clone --branch v1.0 --depth 1 https://gitlab.potapov.me/courses/fastapi_pro.git
cd fastapi_pro
uv sync --group test
uv run --group test pytest graphql/tests

#Чеклист GraphQL

Strawberry установлен
Типы определены
Query и Mutation
DataLoader для N+1
Аутентификация через context
Обработка ошибок

Критерий

REST

GraphQL

Запросы

Фиксированные endpoints

Гибкий запрос

Over-fetching

Получаешь лишние данные

Только нужные поля

Under-fetching

Несколько запросов

Один запрос

Версионирование

/api/v1, /api/v2

Эволюция схемы