API REST/GraphQL

Budowanie API to miejsce, gdzie backend-development spotyka się z prawdziwym światem. Niezależnie od tego, czy tworzysz prosty endpoint REST, czy złożony schemat GraphQL, Claude Code przekształca tworzenie API z żmudnego boilerplate’u w eleganckie, dobrze przetestowane usługi. Ta lekcja pokazuje, jak wykorzystać wsparcie AI do projektowania, implementacji i utrzymania API.

Rewolucja w tworzeniu API

Scenariusz: Twój startup potrzebuje API obsługującego aplikacje mobilne, klientów internetowych i integracje z usługami trzecimi. Wymagania obejmują aktualizacje w czasie rzeczywistym, złożoną autoryzację, rate limiting i wsparcie dla 100K dziennie aktywnych użytkowników. Tradycyjne podejście: miesiące rozwoju. Z Claude Code: tygodnie.

Tradycyjne vs rozwój API oparty na AI

Tradycyjne podejście

Tydzień 1-2: Projektowanie API
- Ręczne pisanie specyfikacji OpenAPI
- Debata REST vs GraphQL
- Projektowanie przepływu autentykacji

Tydzień 3-4: Implementacja
- Pisanie boilerplate dla każdego endpointu
- Ręczna implementacja walidacji
- Dodanie obsługi błędów wszędzie

Tydzień 5-6: Testowanie i dokumentacja
- Pisanie testów dla każdego endpointu
- Tworzenie kolekcji Postman
- Dokumentowanie każdego parametru

Tydzień 7+: Udoskonalenia
- Naprawianie niespójności
- Dodawanie brakujących funkcji
- Optymalizacja wydajności

Podejście Claude Code

Dzień 1-2: Projektowanie i implementacja
> Zaprojektuj RESTful API dla naszej platformy e-commerce
> Uwzględnij autentykację, produkty, zamówienia i użytkowników
> Przestrzegaj najlepszych praktyk nazewnictwa i struktury

> Zaimplementuj API używając FastAPI z:
> - Autentykacją JWT
> - Walidacją wejściową
> - Obsługą błędów
> - Rate limitingiem

Dzień 3: Testowanie i dokumentacja
> Wygeneruj kompleksowe testy dla wszystkich endpointów
> Utwórz dokumentację OpenAPI
> Dodaj przykłady żądań i odpowiedzi

Dzień 4-5: Dopracowanie i wdrożenie
> Dodaj strategię cache'owania
> Zaimplementuj paginację
> Skonfiguruj monitoring

Tworzenie RESTful API

Projektowanie REST API

> Zaprojektuj RESTful API dla systemu zarządzania zadaniami z:
> - Autentykacją i autoryzacją użytkowników
> - Projektami, zadaniami i komentarzami
> - Załącznikami plików
> - Powiadomieniami w czasie rzeczywistym
> Przestrzegaj najlepszych praktyk i konwencji REST

Claude generuje kompleksowy projekt API:

Struktura zasobów

# Zasoby i endpointy API

## Autentykacja
POST   /auth/register
POST   /auth/login
POST   /auth/refresh
POST   /auth/logout

## Użytkownicy
GET    /users              # Lista użytkowników (tylko admin)
GET    /users/{id}         # Profil użytkownika
PUT    /users/{id}         # Aktualizacja użytkownika
DELETE /users/{id}         # Usunięcie konta

## Projekty
GET    /projects           # Lista projektów użytkownika
POST   /projects           # Tworzenie projektu
GET    /projects/{id}      # Szczegóły projektu
PUT    /projects/{id}      # Aktualizacja projektu
DELETE /projects/{id}      # Usunięcie projektu

## Zadania
GET    /projects/{pid}/tasks      # Lista zadań
POST   /projects/{pid}/tasks      # Tworzenie zadania
GET    /tasks/{id}               # Pobieranie zadania
PUT    /tasks/{id}               # Aktualizacja zadania
DELETE /tasks/{id}               # Usunięcie zadania
PATCH  /tasks/{id}/status        # Aktualizacja tylko statusu

## Komentarze
GET    /tasks/{tid}/comments      # Lista komentarzy
POST   /tasks/{tid}/comments      # Dodanie komentarza
PUT    /comments/{id}            # Edycja komentarza
DELETE /comments/{id}            # Usunięcie komentarza

Konwencje HTTP

## Kody statusu
200 OK              - Udany GET/PUT
201 Created         - Udany POST
204 No Content      - Udany DELETE
400 Bad Request     - Nieprawidłowe dane
401 Unauthorized    - Brak/nieprawidłowa auth
403 Forbidden       - Brak uprawnień
404 Not Found       - Zasób nie istnieje
409 Conflict        - Duplikat/konflikt
429 Too Many Requests - Ograniczenie częstotliwości

## Nagłówki
Authorization: Bearer <token>
Content-Type: application/json
X-Request-ID: <uuid>
X-RateLimit-Remaining: 100
X-RateLimit-Reset: 1642345678

## Parametry zapytania
?page=1&limit=20           # Paginacja
?sort=created_at&order=desc # Sortowanie
?filter[status]=active     # Filtrowanie
?include=user,project      # Relacje
?fields=id,title,status    # Wybrane pola

Implementacja z FastAPI

> Zaimplementuj API zarządzania zadaniami używając FastAPI:
> - Użyj SQLAlchemy dla bazy danych
> - Pydantic do walidacji
> - JWT do autentykacji
> - Uwzględnij middleware dla CORS i rate limiting

Claude generuje kompletną implementację:

Główna aplikacja

main.py

from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
import uvicorn

from app.core.config import settings
from app.core.security import get_current_user
from app.api import auth, users, projects, tasks
from app.db.base import engine, Base
from app.middleware.rate_limit import RateLimitMiddleware

# Tworzenie tabel bazy danych
Base.metadata.create_all(bind=engine)

# Inicjalizacja aplikacji FastAPI
app = FastAPI(
    title="API zarządzania zadaniami",
    description="RESTful API do zarządzania zadaniami",
    version="1.0.0",
    docs_url="/api/docs",
    redoc_url="/api/redoc"
)

# Dodanie middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.add_middleware(
    RateLimitMiddleware,
    calls=100,
    period=3600  # 100 wywołań na godzinę
)

# Uwzględnienie routerów
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/users", tags=["users"])
app.include_router(projects.router, prefix="/api/projects", tags=["projects"])
app.include_router(tasks.router, prefix="/api/tasks", tags=["tasks"])

@app.get("/api/health")
async def health_check():
    return {"status": "healthy"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Autentykacja

api/auth.py

from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from app.core.security import create_access_token, verify_password, get_password_hash
from app.core.config import settings
from app.db.session import get_db
from app.models.user import User
from app.schemas.auth import Token, UserCreate, UserLogin

router = APIRouter()

@router.post("/register", response_model=Token)
async def register(
    user_data: UserCreate,
    db: Session = Depends(get_db)
):
    # Sprawdzenie czy użytkownik istnieje
    if db.query(User).filter(User.email == user_data.email).first():
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Email już jest zarejestrowany"
        )

    # Tworzenie nowego użytkownika
    user = User(
        email=user_data.email,
        username=user_data.username,
        hashed_password=get_password_hash(user_data.password)
    )
    db.add(user)
    db.commit()
    db.refresh(user)

    # Generowanie tokenu
    access_token = create_access_token(
        subject=str(user.id),
        expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    )

    return {
        "access_token": access_token,
        "token_type": "bearer",
        "user": user
    }

@router.post("/login", response_model=Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    # Autentykacja użytkownika
    user = db.query(User).filter(User.email == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Nieprawidłowy email lub hasło"
        )

    # Generowanie tokenu
    access_token = create_access_token(subject=str(user.id))

    return {
        "access_token": access_token,
        "token_type": "bearer",
        "user": user
    }

Endpointy zadań

api/tasks.py

from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session

from app.db.session import get_db
from app.models.task import Task
from app.models.user import User
from app.schemas.task import TaskCreate, TaskUpdate, TaskResponse
from app.core.security import get_current_user
from app.core.pagination import paginate

router = APIRouter()

@router.get("/", response_model=List[TaskResponse])
async def list_tasks(
    page: int = Query(1, ge=1),
    limit: int = Query(20, ge=1, le=100),
    status: Optional[str] = None,
    project_id: Optional[int] = None,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    query = db.query(Task).filter(Task.user_id == current_user.id)

    if status:
        query = query.filter(Task.status == status)
    if project_id:
        query = query.filter(Task.project_id == project_id)

    return paginate(query, page, limit)

@router.post("/", response_model=TaskResponse, status_code=201)
async def create_task(
    task_data: TaskCreate,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    # Weryfikacja dostępu do projektu
    project = db.query(Project).filter(
        Project.id == task_data.project_id,
        Project.user_id == current_user.id
    ).first()

    if not project:
        raise HTTPException(
            status_code=404,
            detail="Projekt nie został znaleziony"
        )

    # Tworzenie zadania
    task = Task(
        **task_data.dict(),
        user_id=current_user.id
    )
    db.add(task)
    db.commit()
    db.refresh(task)

    return task

@router.put("/{task_id}", response_model=TaskResponse)
async def update_task(
    task_id: int,
    task_update: TaskUpdate,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    # Pobieranie zadania
    task = db.query(Task).filter(
        Task.id == task_id,
        Task.user_id == current_user.id
    ).first()

    if not task:
        raise HTTPException(status_code=404, detail="Zadanie nie zostało znalezione")

    # Aktualizacja pól
    for field, value in task_update.dict(exclude_unset=True).items():
        setattr(task, field, value)

    db.commit()
    db.refresh(task)

    return task

Zaawansowane funkcje REST

Paginacja i filtrowanie

> Zaimplementuj zaawansowaną paginację i filtrowanie dla endpointu zadań:
> - Paginację opartą na kursorze dla danych w czasie rzeczywistym
> - Złożone filtrowanie z operatorami
> - Wyszukiwanie pełnotekstowe
> - Sortowanie według wielu pól

# Zaawansowana obsługa zapytań
from typing import Optional, List
from fastapi import Query
from sqlalchemy import or_, and_

class TaskQueryParams:
    def __init__(
        self,
        # Paginacja
        cursor: Optional[str] = None,
        limit: int = Query(20, le=100),

        # Filtrowanie
        status: Optional[List[str]] = Query(None),
        priority: Optional[str] = Query(None),
        assigned_to: Optional[int] = None,
        created_after: Optional[datetime] = None,
        created_before: Optional[datetime] = None,

        # Wyszukiwanie
        search: Optional[str] = None,

        # Sortowanie
        sort_by: List[str] = Query(["created_at"]),
        sort_order: List[str] = Query(["desc"])
    ):
        self.cursor = cursor
        self.limit = limit
        self.filters = self._build_filters(locals())
        self.search = search
        self.sorting = list(zip(sort_by, sort_order))

    def apply_to_query(self, query):
        # Zastosowanie filtrów
        if self.filters:
            query = query.filter(and_(*self.filters))

        # Zastosowanie wyszukiwania
        if self.search:
            query = query.filter(
                or_(
                    Task.title.ilike(f"%{self.search}%"),
                    Task.description.ilike(f"%{self.search}%")
                )
            )

        # Zastosowanie sortowania
        for field, order in self.sorting:
            column = getattr(Task, field)
            query = query.order_by(
                column.desc() if order == "desc" else column
            )

        # Zastosowanie paginacji kursorowej
        if self.cursor:
            query = query.filter(Task.id > self.decode_cursor(self.cursor))

        return query.limit(self.limit + 1)

Rate limiting

middleware/rate_limit.py

from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
import time
from collections import defaultdict
import asyncio

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, calls: int = 100, period: int = 3600):
        super().__init__(app)
        self.calls = calls
        self.period = period
        self.clients = defaultdict(list)
        self.cleanup_task = asyncio.create_task(self.cleanup())

    async def dispatch(self, request: Request, call_next):
        # Pobranie identyfikatora klienta
        client_id = request.client.host
        if auth := request.headers.get("Authorization"):
            # Użycie ID użytkownika dla uwierzytelnionych żądań
            client_id = self.get_user_id_from_token(auth)

        # Sprawdzenie limitu częstotliwości
        now = time.time()
        requests = self.clients[client_id]
        requests = [req for req in requests if req > now - self.period]

        if len(requests) >= self.calls:
            raise HTTPException(
                status_code=429,
                detail="Przekroczono limit częstotliwości",
                headers={
                    "X-RateLimit-Limit": str(self.calls),
                    "X-RateLimit-Remaining": "0",
                    "X-RateLimit-Reset": str(int(requests[0] + self.period))
                }
            )

        # Rejestracja żądania
        requests.append(now)
        self.clients[client_id] = requests

        # Dodanie nagłówków limitu częstotliwości
        response = await call_next(request)
        response.headers["X-RateLimit-Limit"] = str(self.calls)
        response.headers["X-RateLimit-Remaining"] = str(self.calls - len(requests))

        return response

Tworzenie GraphQL API

Projektowanie schematów GraphQL

> Zaprojektuj schemat GraphQL dla tego samego systemu zarządzania zadaniami.
> Uwzględnij:
> - Definicje typów dla wszystkich encji
> - Zapytania do odczytu danych
> - Mutacje do modyfikacji
> - Subskrypcje dla aktualizacji w czasie rzeczywistym
> - Niestandardowe skalary dla dat

Claude generuje kompleksowy schemat GraphQL:

schema.graphql

scalar DateTime
scalar Upload

# Typ użytkownika
type User {
  id: ID!
  email: String!
  username: String!
  createdAt: DateTime!
  updatedAt: DateTime!

  # Relacje
  projects(
    first: Int = 10
    after: String
    filter: ProjectFilter
  ): ProjectConnection!

  tasks(
    first: Int = 10
    after: String
    filter: TaskFilter
  ): TaskConnection!
}

# Typ projektu
type Project {
  id: ID!
  title: String!
  description: String
  color: String
  createdAt: DateTime!
  updatedAt: DateTime!

  # Relacje
  owner: User!
  tasks(
    first: Int = 10
    after: String
    filter: TaskFilter
    orderBy: TaskOrderBy
  ): TaskConnection!

  members: [User!]!

  # Pola obliczone
  taskCount: Int!
  completedTaskCount: Int!
  progress: Float!
}

# Typ zadania
type Task {
  id: ID!
  title: String!
  description: String
  status: TaskStatus!
  priority: TaskPriority!
  dueDate: DateTime
  completedAt: DateTime
  createdAt: DateTime!
  updatedAt: DateTime!

  # Relacje
  project: Project!
  assignee: User
  creator: User!
  comments(
    first: Int = 10
    after: String
  ): CommentConnection!

  attachments: [Attachment!]!
  tags: [String!]!
}

# Enumeracje
enum TaskStatus {
  TODO
  IN_PROGRESS
  IN_REVIEW
  COMPLETED
  ARCHIVED
}

enum TaskPriority {
  LOW
  MEDIUM
  HIGH
  URGENT
}

# Typy wejściowe
input CreateTaskInput {
  title: String!
  description: String
  projectId: ID!
  assigneeId: ID
  priority: TaskPriority = MEDIUM
  dueDate: DateTime
  tags: [String!]
}

input UpdateTaskInput {
  title: String
  description: String
  status: TaskStatus
  priority: TaskPriority
  assigneeId: ID
  dueDate: DateTime
  tags: [String!]
}

# Typy paginacji
type PageInfo {
  hasNextPage: Boolean!
  hasPreviousPage: Boolean!
  startCursor: String
  endCursor: String
}

type TaskConnection {
  edges: [TaskEdge!]!
  pageInfo: PageInfo!
  totalCount: Int!
}

type TaskEdge {
  node: Task!
  cursor: String!
}

# Typy główne
type Query {
  # Zapytania użytkowników
  me: User!
  user(id: ID!): User
  users(
    first: Int = 10
    after: String
    search: String
  ): UserConnection!

  # Zapytania projektów
  project(id: ID!): Project
  projects(
    first: Int = 10
    after: String
    filter: ProjectFilter
  ): ProjectConnection!

  # Zapytania zadań
  task(id: ID!): Task
  tasks(
    first: Int = 10
    after: String
    filter: TaskFilter
    orderBy: TaskOrderBy
  ): TaskConnection!

  # Wyszukiwanie
  search(
    query: String!
    types: [SearchType!]
  ): SearchResult!
}

type Mutation {
  # Mutacje autentykacji
  register(input: RegisterInput!): AuthPayload!
  login(email: String!, password: String!): AuthPayload!
  logout: Boolean!
  refreshToken(token: String!): AuthPayload!

  # Mutacje projektów
  createProject(input: CreateProjectInput!): Project!
  updateProject(id: ID!, input: UpdateProjectInput!): Project!
  deleteProject(id: ID!): Boolean!

  # Mutacje zadań
  createTask(input: CreateTaskInput!): Task!
  updateTask(id: ID!, input: UpdateTaskInput!): Task!
  deleteTask(id: ID!): Boolean!

  # Operacje grupowe
  batchUpdateTasks(ids: [ID!]!, input: UpdateTaskInput!): [Task!]!

  # Upload pliku
  uploadAttachment(taskId: ID!, file: Upload!): Attachment!
}

type Subscription {
  # Subskrypcje zadań
  taskCreated(projectId: ID!): Task!
  taskUpdated(id: ID!): Task!
  taskDeleted(id: ID!): ID!

  # Subskrypcje projektów
  projectUpdated(id: ID!): Project!

  # Obecność
  userPresence(projectId: ID!): PresenceEvent!
}

Implementacja ze Strawberry (Python)

> Zaimplementuj GraphQL API używając frameworka Strawberry:
> - Użyj SQLAlchemy dla bazy danych
> - Dodaj DataLoader dla zapobiegania N+1
> - Zaimplementuj autentykację
> - Dodaj wsparcie dla subskrypcji

Definicja schematu

graphql/schema.py

import strawberry
from typing import List, Optional
from datetime import datetime
from strawberry.types import Info

from app.graphql.types import User, Project, Task
from app.graphql.resolvers import (
    get_current_user,
    resolve_projects,
    resolve_tasks
)
from app.graphql.mutations import AuthMutations, TaskMutations

@strawberry.type
class Query:
    @strawberry.field
    async def me(self, info: Info) -> User:
        return await get_current_user(info)

    @strawberry.field
    async def projects(
        self,
        info: Info,
        first: int = 10,
        after: Optional[str] = None
    ) -> ProjectConnection:
        user = await get_current_user(info)
        return await resolve_projects(user, first, after)

    @strawberry.field
    async def tasks(
        self,
        info: Info,
        first: int = 10,
        after: Optional[str] = None,
        status: Optional[TaskStatus] = None,
        project_id: Optional[strawberry.ID] = None
    ) -> TaskConnection:
        user = await get_current_user(info)
        return await resolve_tasks(
            user, first, after, status, project_id
        )

@strawberry.type
class Mutation(AuthMutations, TaskMutations):
    pass

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def task_created(
        self, info: Info, project_id: strawberry.ID
    ) -> Task:
        async for task in task_created_generator(project_id):
            yield task

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

Konfiguracja DataLoader

graphql/dataloaders.py

from strawberry.dataloader import DataLoader
from typing import List
from sqlalchemy.orm import Session

class UserLoader(DataLoader):
    def __init__(self, db: Session):
        super().__init__(load_fn=self.load_users)
        self.db = db

    async def load_users(self, user_ids: List[int]) -> List[User]:
        users = self.db.query(User).filter(
            User.id.in_(user_ids)
        ).all()

        # Mapowanie w celu zachowania kolejności
        user_map = {user.id: user for user in users}
        return [user_map.get(uid) for uid in user_ids]

class ProjectLoader(DataLoader):
    def __init__(self, db: Session):
        super().__init__(load_fn=self.load_projects)
        self.db = db

    async def load_projects(self, project_ids: List[int]) -> List[Project]:
        projects = self.db.query(Project).filter(
            Project.id.in_(project_ids)
        ).all()

        project_map = {p.id: p for p in projects}
        return [project_map.get(pid) for pid in project_ids]

# Konfiguracja kontekstu
async def get_context(request):
    return {
        "request": request,
        "db": get_db(),
        "user_loader": UserLoader(get_db()),
        "project_loader": ProjectLoader(get_db()),
        "current_user": await get_current_user_from_request(request)
    }

Subskrypcje

graphql/subscriptions.py

import asyncio
from typing import AsyncGenerator
import strawberry
from broadcaster import Broadcast

# Inicjalizacja broadcastera
broadcast = Broadcast("redis://localhost:6379")

@strawberry.type
class TaskCreatedSubscription:
    @strawberry.subscription
    async def task_created(
        self,
        project_id: strawberry.ID
    ) -> AsyncGenerator[Task, None]:
        async with broadcast.subscribe(f"project:{project_id}:tasks") as subscriber:
            async for event in subscriber:
                if event["type"] == "task_created":
                    yield Task.from_dict(event["data"])

# Publikowanie zdarzeń
async def publish_task_created(task: Task):
    await broadcast.publish(
        f"project:{task.project_id}:tasks",
        {
            "type": "task_created",
            "data": task.to_dict()
        }
    )

Strategie testowania API

Testowanie jednostkowe endpointów

> Wygeneruj kompleksowe testy dla naszego API:
> - Testy jednostkowe dla każdego endpointu
> - Testy integracyjne z bazą danych
> - Testy autentykacji/autoryzacji
> - Testy obsługi błędów
> - Testy wydajności

Testy REST API

tests/test_api_tasks.py

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from app.main import app
from app.tests.utils import (
    create_test_user,
    create_test_project,
    create_test_task,
    get_auth_headers
)

client = TestClient(app)

class TestTaskAPI:
    def test_create_task(self, db: Session):
        user = create_test_user(db)
        project = create_test_project(db, user)
        headers = get_auth_headers(user)

        response = client.post(
            "/api/tasks/",
            headers=headers,
            json={
                "title": "Zadanie testowe",
                "description": "Opis testowy",
                "project_id": project.id,
                "priority": "HIGH"
            }
        )

        assert response.status_code == 201
        data = response.json()
        assert data["title"] == "Zadanie testowe"
        assert data["priority"] == "HIGH"
        assert data["status"] == "TODO"

    def test_list_tasks_pagination(self, db: Session):
        user = create_test_user(db)
        project = create_test_project(db, user)

        # Tworzenie 25 zadań
        for i in range(25):
            create_test_task(db, project, f"Zadanie {i}")

        headers = get_auth_headers(user)

        # Pierwsza strona
        response = client.get(
            "/api/tasks/?limit=10&page=1",
            headers=headers
        )
        assert response.status_code == 200
        data = response.json()
        assert len(data["items"]) == 10
        assert data["total"] == 25
        assert data["page"] == 1

        # Druga strona
        response = client.get(
            "/api/tasks/?limit=10&page=2",
            headers=headers
        )
        assert len(response.json()["items"]) == 10

    def test_update_task_authorization(self, db: Session):
        user1 = create_test_user(db, "user1@test.com")
        user2 = create_test_user(db, "user2@test.com")
        project = create_test_project(db, user1)
        task = create_test_task(db, project)

        # Próba aktualizacji zadania innego użytkownika
        headers = get_auth_headers(user2)
        response = client.put(
            f"/api/tasks/{task.id}",
            headers=headers,
            json={"title": "Zhakowane!"}
        )

        assert response.status_code == 404  # Nie znaleziono dla nieautoryzowanego użytkownika

    @pytest.mark.parametrize("invalid_data", [
        {"title": ""},  # Pusty tytuł
        {"title": "A" * 256},  # Za długi
        {"priority": "INVALID"},  # Nieprawidłowy enum
        {"project_id": 999999}  # Nieistniejący projekt
    ])
    def test_create_task_validation(self, db: Session, invalid_data):
        user = create_test_user(db)
        headers = get_auth_headers(user)

        response = client.post(
            "/api/tasks/",
            headers=headers,
            json=invalid_data
        )

        assert response.status_code == 422  # Błąd walidacji

Testy GraphQL

tests/test_graphql.py

import pytest
from strawberry.test import GraphQLTestClient

from app.graphql.schema import schema
from app.tests.utils import create_test_user, create_test_project

class TestGraphQLAPI:
    def test_query_projects(self, client: GraphQLTestClient):
        query = """
        query GetProjects($first: Int!) {
            projects(first: $first) {
                edges {
                    node {
                        id
                        title
                        taskCount
                    }
                }
                pageInfo {
                    hasNextPage
                    endCursor
                }
            }
        }
        """

        result = client.execute(
            query,
            variables={"first": 10},
            context_value={"user": create_test_user()}
        )

        assert not result.errors
        assert "projects" in result.data

    def test_create_task_mutation(self, client: GraphQLTestClient):
        mutation = """
        mutation CreateTask($input: CreateTaskInput!) {
            createTask(input: $input) {
                id
                title
                status
                project {
                    id
                    title
                }
            }
        }
        """

        user = create_test_user()
        project = create_test_project(user)

        result = client.execute(
            mutation,
            variables={
                "input": {
                    "title": "Nowe zadanie",
                    "projectId": str(project.id),
                    "priority": "HIGH"
                }
            },
            context_value={"user": user}
        )

        assert not result.errors
        task = result.data["createTask"]
        assert task["title"] == "Nowe zadanie"
        assert task["status"] == "TODO"

    def test_subscription(self, client: GraphQLTestClient):
        subscription = """
        subscription OnTaskCreated($projectId: ID!) {
            taskCreated(projectId: $projectId) {
                id
                title
                creator {
                    username
                }
            }
        }
        """

        # Test konfiguracji subskrypcji
        result = client.execute(
            subscription,
            variables={"projectId": "1"}
        )

        # W prawdziwym teście, wywołałby tworzenie zadania
        # i sprawdził czy subskrypcja otrzymuje zdarzenie

Testowanie wydajności

tests/test_performance.py

import asyncio
import time
from locust import HttpUser, task, between

class APIUser(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        # Logowanie w celu otrzymania tokenu
        response = self.client.post("/api/auth/login", json={
            "username": "test@example.com",
            "password": "testpass"
        })
        self.token = response.json()["access_token"]
        self.headers = {"Authorization": f"Bearer {self.token}"}

    @task(3)
    def list_tasks(self):
        self.client.get("/api/tasks/", headers=self.headers)

    @task(2)
    def get_project(self):
        self.client.get("/api/projects/1", headers=self.headers)

    @task(1)
    def create_task(self):
        self.client.post(
            "/api/tasks/",
            headers=self.headers,
            json={
                "title": f"Zadanie {time.time()}",
                "project_id": 1
            }
        )

# Uruchom z: locust -f test_performance.py --host=http://localhost:8000

Najlepsze praktyki bezpieczeństwa API

Autentykacja i autoryzacja

> Zaimplementuj kompleksowe bezpieczeństwo dla naszego API:
> - JWT z tokenami odświeżania
> - Integrację OAuth2
> - Autentykację kluczem API dla usług
> - Kontrolę dostępu opartą na rolach
> - Rate limiting na użytkownika/IP

core/security.py

from datetime import datetime, timedelta
from typing import Optional, Union
import jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")

class SecurityManager:
    def __init__(self, secret_key: str, algorithm: str = "HS256"):
        self.secret_key = secret_key
        self.algorithm = algorithm

    def create_access_token(
        self,
        subject: Union[str, int],
        expires_delta: Optional[timedelta] = None,
        additional_claims: dict = None
    ) -> str:
        expire = datetime.utcnow() + (
            expires_delta or timedelta(minutes=15)
        )

        to_encode = {
            "sub": str(subject),
            "exp": expire,
            "iat": datetime.utcnow(),
            "type": "access"
        }

        if additional_claims:
            to_encode.update(additional_claims)

        return jwt.encode(to_encode, self.secret_key, self.algorithm)

    def create_refresh_token(self, subject: Union[str, int]) -> str:
        expire = datetime.utcnow() + timedelta(days=7)
        to_encode = {
            "sub": str(subject),
            "exp": expire,
            "type": "refresh"
        }
        return jwt.encode(to_encode, self.secret_key, self.algorithm)

    def decode_token(self, token: str) -> dict:
        try:
            payload = jwt.decode(
                token, self.secret_key, algorithms=[self.algorithm]
            )
            return payload
        except jwt.ExpiredSignatureError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Token wygasł"
            )
        except jwt.JWTError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Nie można zweryfikować poświadczeń"
            )

# System uprawnień
class Permission:
    def __init__(self, resource: str, action: str):
        self.resource = resource
        self.action = action

    def check(self, user: User) -> bool:
        # Sprawdzenie czy użytkownik ma uprawnienie
        return self in user.get_permissions()

class RequirePermission:
    def __init__(self, permission: Permission):
        self.permission = permission

    def __call__(self, current_user: User = Depends(get_current_user)):
        if not self.permission.check(current_user):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Niewystarczające uprawnienia"
            )
        return current_user

# Użycie w endpointach
@router.delete(
    "/{task_id}",
    dependencies=[Depends(RequirePermission(Permission("task", "delete")))]
)
async def delete_task(task_id: int):
    pass

Walidacja i sanityzacja danych wejściowych

schemas/validators.py

from pydantic import BaseModel, validator, constr, conint
import bleach
from typing import Optional

class SanitizedStr(str):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v):
        if not isinstance(v, str):
            raise TypeError("Wymagany ciąg znaków")
        # Usunięcie niebezpiecznego HTML/skryptów
        return bleach.clean(v, tags=[], strip=True)

class TaskCreateSchema(BaseModel):
    title: constr(min_length=1, max_length=200)
    description: Optional[SanitizedStr] = None
    priority: conint(ge=1, le=4)

    @validator("title")
    def validate_title(cls, v):
        if not v.strip():
            raise ValueError("Tytuł nie może być pusty")
        return v.strip()

    class Config:
        # Zapobieganie dodatkowym polom
        extra = "forbid"

Dokumentacja i wersjonowanie API

Dokumentacja OpenAPI

> Wygeneruj kompleksową dokumentację API:
> - Specyfikację OpenAPI/Swagger
> - Przykłady żądań/odpowiedzi
> - Szczegóły autentykacji
> - Katalog odpowiedzi błędów
> - Strategię wersjonowania

# main.py - Rozszerzona dokumentacja
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI()

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title="API zarządzania zadaniami",
        version="2.0.0",
        description="""
        ## Przegląd
        RESTful API do zarządzania zadaniami z aktualizacjami w czasie rzeczywistym.

        ## Autentykacja
        Używa tokenów JWT Bearer. Uwzględnij w nagłówku Authorization:

            Authorization: Bearer <token>

        ## Ograniczenie częstotliwości
        - 100 żądań na godzinę dla uwierzytelnionych użytkowników
        - 20 żądań na godzinę dla nieuwierzytelnionych

        ## Błędy
        Standardowe kody statusu HTTP. Odpowiedzi błędów zawierają:

            {
                "detail": "Opis błędu",
                "code": "KOD_BŁĘDU",
                "field": "nazwa_pola"  // Dla błędów walidacji
            }
        """,
        routes=app.routes,
    )

    # Dodanie schematu bezpieczeństwa
    openapi_schema["components"]["securitySchemes"] = {
        "BearerAuth": {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "JWT",
        }
    }

    # Dodanie przykładów
    openapi_schema["paths"]["/api/tasks/"]["post"]["requestBody"]["content"]["application/json"]["examples"] = {
        "simple": {
            "summary": "Proste zadanie",
            "value": {
                "title": "Przegląd PR",
                "project_id": 1
            }
        },
        "complete": {
            "summary": "Kompletne zadanie",
            "value": {
                "title": "Implementacja funkcji",
                "description": "Dodanie autentykacji użytkownika",
                "priority": 3,
                "due_date": "2024-12-31T23:59:59Z",
                "assigned_to": 5
            }
        }
    }

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

Wersjonowanie API

api/v1/routes.py

from fastapi import APIRouter

v1_router = APIRouter(prefix="/api/v1")

# api/v2/routes.py
v2_router = APIRouter(prefix="/api/v2")

# main.py
app.include_router(v1_router)
app.include_router(v2_router)

# Negocjacja wersji przez nagłówki
@app.middleware("http")
async def api_version_middleware(request: Request, call_next):
    version = request.headers.get("API-Version", "v2")
    request.state.api_version = version
    response = await call_next(request)
    response.headers["API-Version"] = version
    return response

Powiązane lekcje

Operacje bazy danych Zbuduj warstwę danych, która zasila twoje API

Systemy autentykacji Zabezpiecz swoje API z solidną autentykacją

Testowanie API Upewnij się, że twoje API działa niezawodnie

Następne kroki

Nauczyłeś się wykorzystywać Claude Code do szybkiego rozwoju API, od projektowania przez implementację po testowanie. Kluczem jest pozwolenie Claude’owi obsłużyć boilerplate, podczas gdy ty skupiasz się na logice biznesowej i decyzjach architektonicznych.

Pamiętaj: świetne API to coś więcej niż tylko endpointy - to spójność, bezpieczeństwo, wydajność i doświadczenie developera. Użyj Claude Code do egzekwowania najlepszych praktyk we wszystkich tych wymiarach, tworząc API, które są przyjemnością w budowaniu i używaniu.