W tym przewodniku poznasz, jak stworzyć kompletne REST API w FastAPI: od podstawowych endpointów po zaawansowane funkcjonalności, walidację danych, obsługę błędów, testy i automatyczną dokumentację.
Podstawy tworzenia REST API
Tworzenie API w FastAPI opiera się na kilku prostych krokach:
- Zdefiniowanie endpointów (
@app.get(),@app.post()itd.) - Walidacja danych przy pomocy Pydantic
- Zwracanie odpowiedzi w formacie JSON
- Automatyczna dokumentacja pod
/docsi/redoc
Instalacja
Bash1pip install fastapi uvicorn[standard]
Przykład najprostszej aplikacji
Python1 2 3 4 5 6 7 8 9 10 11from fastapi import FastAPI app = FastAPI(title="My API", version="1.0.0") @app.get("/") def read_root(): return {"message": "Witaj w moim API"} @app.get("/health") def health_check(): return {"status": "ok"}
Uruchom serwer:
Bash1uvicorn main:app --reload
Po uruchomieniu serwera możesz wejść na http://127.0.0.1:8000/docs — dokumentacja Swaggera jest generowana automatycznie. Alternatywnie możesz użyć http://127.0.0.1:8000/redoc dla dokumentacji w stylu ReDoc.
CRUD – Create, Read, Update, Delete
Najczęstszy wzorzec w REST API to CRUD, czyli cztery operacje: tworzenie, odczyt, aktualizacja i usuwanie danych.
Przykład API użytkowników
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80from fastapi import FastAPI, HTTPException, status from pydantic import BaseModel, EmailStr from typing import List, Optional app = FastAPI() class User(BaseModel): id: int name: str email: EmailStr class UserCreate(BaseModel): name: str email: EmailStr # Tymczasowa "baza danych" w pamięci users: List[User] = [] @app.post("/users", status_code=status.HTTP_201_CREATED, response_model=User) def create_user(user: UserCreate): """Utwórz nowego użytkownika""" new_id = max([u.id for u in users], default=0) + 1 new_user = User(id=new_id, **user.dict()) users.append(new_user) return new_user @app.get("/users", response_model=List[User]) def get_users(): """Pobierz listę wszystkich użytkowników""" return users @app.get("/users/{user_id}", response_model=User) def get_user(user_id: int): """Pobierz użytkownika po ID""" for u in users: if u.id == user_id: return u raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with id {user_id} not found" ) @app.put("/users/{user_id}", response_model=User) def update_user(user_id: int, updated_user: UserCreate): """Zaktualizuj użytkownika (pełna aktualizacja)""" for i, u in enumerate(users): if u.id == user_id: users[i] = User(id=user_id, **updated_user.dict()) return users[i] raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with id {user_id} not found" ) @app.patch("/users/{user_id}", response_model=User) def partial_update_user(user_id: int, updated_data: dict): """Zaktualizuj część danych użytkownika""" for i, u in enumerate(users): if u.id == user_id: updated_dict = u.dict() updated_dict.update(updated_data) users[i] = User(**updated_dict) return users[i] raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with id {user_id} not found" ) @app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_user(user_id: int): """Usuń użytkownika""" global users user_exists = any(u.id == user_id for u in users) if not user_exists: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with id {user_id} not found" ) users = [u for u in users if u.id != user_id] return None
Każdy endpoint jest zdefiniowany w sposób czytelny, deklaratywny i typowany — FastAPI automatycznie waliduje dane wejściowe i generuje dokumentację.
Walidacja danych z Pydantic
FastAPI wykorzystuje Pydantic do walidacji danych — każde pole ma typ i reguły, które są automatycznie sprawdzane.
Podstawowa walidacja
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23from pydantic import BaseModel, EmailStr, Field, validator class User(BaseModel): id: int = Field(gt=0, description="ID użytkownika musi być większe od 0") name: str = Field(min_length=3, max_length=100, description="Imię użytkownika") email: EmailStr = Field(..., description="Adres e-mail użytkownika") age: Optional[int] = Field(None, ge=0, le=120, description="Wiek użytkownika") @validator('name') def name_must_not_be_empty(cls, v): if not v.strip(): raise ValueError('Name cannot be empty or only whitespace') return v.strip() class Config: schema_extra = { "example": { "id": 1, "name": "Kacper Sieradziński", "email": "kacper@example.com", "age": 25 } }
Zaawansowana walidacja
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24from pydantic import BaseModel, validator, root_validator from typing import Optional from datetime import datetime class UserCreate(BaseModel): name: str = Field(..., min_length=3, max_length=100) email: EmailStr password: str = Field(..., min_length=8, regex="^(?=.*[A-Za-z])(?=.*\d)") confirm_password: str @validator('email') def email_must_be_valid(cls, v): # EmailStr już waliduje format, ale możesz dodać dodatkową logikę if 'example.com' in v: raise ValueError('Email z domeną example.com nie jest dozwolony') return v @root_validator def passwords_match(cls, values): password = values.get('password') confirm_password = values.get('confirm_password') if password != confirm_password: raise ValueError('Passwords do not match') return values
Jeśli użytkownik prześle błędne dane, FastAPI automatycznie zwróci błąd walidacji:
JSON1 2 3 4 5 6 7 8 9 10 11 12 13 14{ "detail": [ { "loc": ["body", "email"], "msg": "value is not a valid email address", "type": "value_error.email" }, { "loc": ["body", "name"], "msg": "ensure this value has at least 3 characters", "type": "value_error.any_str.min_length" } ] }
Nie trzeba pisać żadnych ręcznych walidatorów — wszystko działa "z pudełka".
Obsługa błędów
FastAPI udostępnia wbudowany system wyjątków (HTTPException) oraz możliwość definiowania globalnych handlerów błędów.
Podstawowa obsługa błędów
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42from fastapi import FastAPI, HTTPException, status from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError from starlette.exceptions import HTTPException as StarletteHTTPException app = FastAPI() @app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc: RequestValidationError): """Globalny handler dla błędów walidacji""" return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={ "status": "error", "message": "Validation error", "details": exc.errors() } ) @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request, exc: StarletteHTTPException): """Globalny handler dla błędów HTTP""" return JSONResponse( status_code=exc.status_code, content={ "status": "error", "message": exc.detail, "path": str(request.url.path) } ) @app.exception_handler(Exception) async def general_exception_handler(request, exc: Exception): """Handler dla nieoczekiwanych błędów""" return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={ "status": "error", "message": "Internal server error", "detail": str(exc) if app.debug else "An unexpected error occurred" } )
Niestandardowe wyjątki
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15from fastapi import HTTPException, status class UserNotFoundError(HTTPException): def __init__(self, user_id: int): super().__init__( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with id {user_id} not found" ) @app.get("/users/{user_id}") def get_user(user_id: int): user = find_user_by_id(user_id) if not user: raise UserNotFoundError(user_id) return user
Dzięki globalnym handlerom błędów można centralnie zarządzać odpowiedziami w całym API.
Asynchroniczność i wydajność
FastAPI jest zbudowane na ASGI (Asynchronous Server Gateway Interface) i pozwala korzystać z async/await, co daje wzrost wydajności przy obsłudze wielu żądań jednocześnie.
Synchroniczny vs asynchroniczny endpoint
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24import asyncio import httpx # Synchroniczny endpoint (blokuje wątek) @app.get("/sync-data") def get_sync_data(): # Symulacja kosztownej operacji I/O import time time.sleep(2) return {"status": "ready"} # Asynchroniczny endpoint (nie blokuje) @app.get("/async-data") async def get_async_data(): # Asynchroniczna operacja I/O await asyncio.sleep(2) return {"status": "ready"} # Asynchroniczne wywołanie zewnętrznego API @app.get("/external-data") async def get_external_data(): async with httpx.AsyncClient() as client: response = await client.get("https://api.example.com/data") return response.json()
Równoległe wykonywanie zadań
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22import asyncio @app.get("/aggregated-data") async def get_aggregated_data(): async def fetch_user_data(): await asyncio.sleep(1) return {"users": [...]} async def fetch_order_data(): await asyncio.sleep(1) return {"orders": [...]} # Wykonaj oba równolegle user_data, order_data = await asyncio.gather( fetch_user_data(), fetch_order_data() ) return { "users": user_data["users"], "orders": order_data["orders"] }
Każde żądanie może być obsługiwane asynchronicznie bez blokowania głównego wątku, co pozwala na obsługę tysięcy równoczesnych połączeń.
Dokumentacja OpenAPI i Swagger
Jedną z największych zalet FastAPI jest automatyczna dokumentacja API. Dostępne są dwa interfejsy:
/docs– Swagger UI (interaktywny, można testować endpointy)/redoc– ReDoc (czytelny, statyczny, lepszy do czytania)
Konfiguracja dokumentacji
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32from fastapi import FastAPI from fastapi.openapi.utils import get_openapi app = FastAPI( title="Users API", version="1.0.0", description="API do zarządzania użytkownikami", terms_of_service="http://example.com/terms/", contact={ "name": "Kacper Sieradziński", "email": "kacper@example.com", }, license_info={ "name": "MIT", "url": "https://opensource.org/licenses/MIT", }, ) # Niestandardowy schemat OpenAPI def custom_openapi(): if app.openapi_schema: return app.openapi_schema openapi_schema = get_openapi( title="Users API", version="1.0.0", description="Custom API documentation", routes=app.routes, ) app.openapi_schema = openapi_schema return app.openapi_schema app.openapi = custom_openapi
Dodawanie opisów do endpointów
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15@app.post( "/users", status_code=201, summary="Utwórz użytkownika", description="Utworzy nowego użytkownika w systemie", response_description="Utworzony użytkownik", tags=["Users"] ) def create_user(user: UserCreate): """Utwórz nowego użytkownika. - **name**: Imię użytkownika (min. 3 znaki) - **email**: Adres e-mail użytkownika (walidacja formatu) """ # ...
Dokumentacja jest generowana automatycznie na podstawie typów danych i adnotacji w kodzie. Dzięki temu klient (frontend, mobile, integracje) zawsze wie, jak wygląda Twoje API.
Testowanie API
FastAPI ma wsparcie dla testów dzięki pytest i klientowi testowemu (TestClient).
Podstawowe testy
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39from fastapi.testclient import TestClient from main import app client = TestClient(app) def test_read_root(): response = client.get("/") assert response.status_code == 200 assert response.json() == {"message": "Witaj w moim API"} def test_create_user(): response = client.post( "/users", json={"name": "Kacper", "email": "kacper@example.com"} ) assert response.status_code == 201 data = response.json() assert data["name"] == "Kacper" assert "id" in data def test_get_user_not_found(): response = client.get("/users/999") assert response.status_code == 404 def test_update_user(): # Najpierw utwórz użytkownika create_response = client.post( "/users", json={"name": "Kacper", "email": "kacper@example.com"} ) user_id = create_response.json()["id"] # Zaktualizuj użytkownika update_response = client.put( f"/users/{user_id}", json={"name": "Kacper S.", "email": "kacper.s@example.com"} ) assert update_response.status_code == 200 assert update_response.json()["name"] == "Kacper S."
Asynchroniczne testy
Python1 2 3 4 5 6 7 8 9import pytest from httpx import AsyncClient @pytest.mark.asyncio async def test_async_endpoint(): async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/async-data") assert response.status_code == 200 assert response.json()["status"] == "ready"
Testy z bazą danych
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from fastapi.testclient import TestClient # Testowa baza danych SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db" engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}) TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) def override_get_db(): try: db = TestingSessionLocal() yield db finally: db.close() app.dependency_overrides[get_db] = override_get_db client = TestClient(app) def test_create_user_with_db(): response = client.post( "/users", json={"name": "Test User", "email": "test@example.com"} ) assert response.status_code == 201
Testy uruchomisz komendą:
Bash1pytest -v
Dependency Injection
FastAPI ma system dependency injection, który pozwala na wstrzykiwanie zależności (np. bazy danych, konfiguracji, tokenów JWT) do endpointów bez duplikowania kodu.
Podstawowe dependency
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14from fastapi import Depends, FastAPI from sqlalchemy.orm import Session def get_db(): db = SessionLocal() try: yield db finally: db.close() @app.get("/users") def list_users(db: Session = Depends(get_db)): users = db.query(User).all() return users
Zagnieżdżone dependencies
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16from fastapi import Depends, Header, HTTPException def verify_token(authorization: str = Header(None)): if not authorization: raise HTTPException(status_code=401, detail="Missing token") # Weryfikacja tokena return authorization def get_current_user(token: str = Depends(verify_token), db: Session = Depends(get_db)): # Dekoduj token i zwróć użytkownika user = decode_token_and_get_user(token, db) return user @app.get("/profile") def get_profile(current_user: User = Depends(get_current_user)): return current_user
Dependency z parametrami
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15from typing import Optional from fastapi import Query, Depends def pagination_params( page: int = Query(1, ge=1), limit: int = Query(10, ge=1, le=100) ): return {"skip": (page - 1) * limit, "limit": limit} @app.get("/users") def list_users(pagination: dict = Depends(pagination_params)): skip = pagination["skip"] limit = pagination["limit"] # Zwróć użytkowników z paginacją return users[skip:skip + limit]
Dzięki dependency injection logika aplikacji pozostaje czysta, modułowa i testowalna.
Obsługa błędów walidacji i kodów statusu
FastAPI automatycznie zwraca odpowiednie statusy (422 Unprocessable Entity dla błędów walidacji), ale możesz również definiować własne kody zwrotne:
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23from fastapi import FastAPI, HTTPException, status @app.post( "/users", status_code=status.HTTP_201_CREATED, responses={ 201: {"description": "User created successfully"}, 400: {"description": "Bad Request - Invalid data"}, 409: {"description": "Conflict - User already exists"} } ) def create_user(user: UserCreate): if user_exists(user.email): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="User with this email already exists" ) if user.id < 0: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid ID" ) return create_user_in_db(user)
Niestandardowe kody statusu w odpowiedziach
Python1 2 3 4 5 6 7 8from fastapi.responses import JSONResponse @app.get("/custom-response") def custom_response(): return JSONResponse( status_code=201, content={"message": "Created", "data": {...}} )
Najlepsze praktyki
Poniżej znajdziesz zestaw najlepszych praktyk do tworzenia REST API w FastAPI:
1. Zawsze używaj modeli Pydantic do walidacji wejść i wyjść
Python1 2 3 4 5 6 7 8 9# Dobrze @app.post("/users", response_model=User) def create_user(user: UserCreate): ... # Źle - brak walidacji @app.post("/users") def create_user(user: dict): ...
2. Oddziel warstwę logiki od warstwy API
Struktura projektu:
Bash1 2 3 4 5 6 7 8 9 10 11 12app/ ├── main.py # Konfiguracja FastAPI ├── routers/ # Endpointy (API layer) │ ├── users.py │ └── orders.py ├── schemas/ # Modele Pydantic │ └── user.py ├── services/ # Logika biznesowa │ └── user_service.py ├── models/ # Modele bazy danych │ └── user.py └── database.py # Konfiguracja bazy danych
3. Wersjonuj API
Python1 2 3 4 5 6 7 8 9 10 11 12 13 14 15from fastapi import APIRouter v1_router = APIRouter(prefix="/api/v1", tags=["v1"]) v2_router = APIRouter(prefix="/api/v2", tags=["v2"]) @v1_router.get("/users") def get_users_v1(): ... @v2_router.get("/users") def get_users_v2(): ... app.include_router(v1_router) app.include_router(v2_router)
4. Używaj asynchronicznych ORM-ów
- SQLAlchemy 2.0 async
- Tortoise ORM
- Databases (encode/databases)
5. Zadbaj o testy jednostkowe i integracyjne
Pisz testy dla każdego endpointu i logiki biznesowej.
6. Zabezpiecz endpointy przez JWT, OAuth2 i CORS
Python1 2 3 4 5 6 7 8 9from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["https://example.com"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )
7. Dokumentuj każde endpointy
Dodawaj opisy, przykłady i tagi do każdego endpointu — dobra dokumentacja = mniej pytań od zespołu frontendowego.
8. Mierz wydajność
Bash1 2 3 4 5# Więcej workerów dla większej przepustowości uvicorn main:app --workers 4 # Z monitorowaniem uvicorn main:app --workers 4 --log-level info
9. Utrzymuj spójne formaty odpowiedzi
Python1 2 3 4 5 6 7 8 9 10 11 12def success_response(data, status_code=200): return { "status": "success", "data": data }, status_code def error_response(message, code=400): return { "status": "error", "message": message, "code": code }, code
10. Korzystaj z dependency injection
Używaj Depends() do wstrzykiwania zależności — czyści kod i redukuje powtórzenia.
Podsumowanie
FastAPI to framework, który łączy prostotę Flask-a z wydajnością Node.js i typowaniem jak w TypeScript. Dzięki niemu możesz w krótkim czasie stworzyć pełnoprawne REST API z automatyczną dokumentacją, testami, walidacją i asynchronicznością.
Najważniejsze zalety FastAPI:
- Automatyczna walidacja — dzięki Pydantic
- Automatyczna dokumentacja — OpenAPI/Swagger "z pudełka"
- Asynchroniczność — wysoka wydajność dzięki
async/await - Typowanie — autouzupełnianie i wykrywanie błędów w IDE
- Prostota — czytelny i deklaratywny kod
Jeśli szukasz frameworka, który pozwoli Ci pisać czysty, szybki i bezpieczny kod backendowy — FastAPI to najlepszy wybór. Zacznij od prostych endpointów, stopniowo dodawaj walidację, obsługę błędów i testy, a zobaczysz, jak szybko możesz stworzyć profesjonalne API.
