Jak zacząć z FastAPI: szybkie API w Pythonie dla początkujących

Dlaczego FastAPI stało się tak popularne i czym właściwie jest

Intuicyjne spojrzenie na FastAPI

FastAPI to framework do budowy interfejsów API w Pythonie, który maksymalnie wykorzystuje typy statyczne znane z adnotacji typu name: str czy age: int. Zamiast traktować je jedynie jako podpowiedź dla programisty lub IDE, FastAPI zamienia je w realne zasady walidacji danych i generowania dokumentacji. Dzięki temu backend przestaje być „czarną skrzynką”, a API staje się jasno opisanym kontraktem między serwerem a klientem.

Najważniejsza myśl: piszesz normalny kod Pythona z typami, a FastAPI sam dba o sprawdzanie danych wejściowych, czytelne błędy dla klienta oraz aktualną dokumentację. To sprawia, że aplikacje backendowe w Pythonie można budować szybciej, pewniej i z mniejszą liczbą niespodzianek na etapie integracji z frontendem lub innymi usługami.

FastAPI, Flask i Django REST Framework – krótkie porównanie

FastAPI często porównuje się do dwóch gigantów świata Pythona: minimalnego Flaska i rozbudowanego Django REST Framework (DRF). Każdy z nich ma swoje miejsce, ale różnią się filozofią i zakresem tego, co dostajesz „z pudełka”.

FastAPI jest szczególnie mocne tam, gdzie liczy się szybkość tworzenia API oraz jakość kontraktu między backendem a klientami (web, mobile, inne serwisy). W małych i średnich projektach potrafi zastąpić zarówno Flaska, jak i cięższe frameworki, nie zmuszając do przyjęcia z góry określonej architektury całej aplikacji.

Co oznacza „fast” w nazwie FastAPI

Słowo „fast” ma w FastAPI dwa poziomy znaczenia. Pierwszy to wydajność techniczna. Framework jest oparty o ASGI (Asynchronous Server Gateway Interface) i może wykorzystywać asynchroniczne funkcje async def, a więc obsługiwać wiele zapytań jednocześnie, nie blokując się przy operacjach sieciowych czy dyskowych. Pod względem surowej liczby obsługiwanych żądań na sekundę FastAPI znajduje się bardzo wysoko w rankingach frameworków w Pythonie.

Drugi aspekt to szybkość pracy programisty. Dzięki wykorzystaniu typów i Pydantic kod endpointów pozostaje krótki i czytelny, a walidacja czy generowanie dokumentacji dzieją się „przy okazji”. Zwykle ilość kodu „klejącego” (sprawdzanie, czy pola są, czy mają dobry typ, przygotowanie błędów 400) spada o kilkadziesiąt procent względem klasycznych podejść, co w małych zespołach przekłada się na realne godziny zaoszczędzone każdego tygodnia.

Dla jakich projektów FastAPI ma największy sens

FastAPI szczególnie dobrze sprawdza się w kilku scenariuszach:

• Mikroserwisy i małe usługi – pojedyncza funkcja biznesowa, wystawiona jako API dla innych systemów.
• Backendy dla frontendu SPA / mobile – tam, gdzie ważna jest dobra i aktualna dokumentacja oraz stabilność kontraktów.
• API dla modeli machine learning – szybkie wystawienie modelu jako endpoint (np. /predict), często w połączeniu z bibliotekami naukowymi Pythona.
• Prototypy i POC – gdy trzeba w kilka godzin postawić działające API do testów.
• Nowe projekty greenfield – gdy nie ma wymogu używania konkretnego frameworka jak Django.

Przy dużych, monolitycznych aplikacjach z mocno rozbudowanym panelem administracyjnym i gotowymi modułami (np. system uprawnień, CMS) Django może nadal mieć przewagę. Natomiast do czystych API FastAPI zwykle wygrywa prostotą, typowaniem i asynchronicznością.

Jak typy w Pythonie stały się paliwem dla dokumentacji

Jeszcze kilka lat temu adnotacje typów w Pythonie były traktowane jako „dopisek” dla narzędzi statycznej analizy kodu. FastAPI pokazało, że można zrobić krok dalej: jeśli funkcja endpointu ma sygnaturę def create_user(user: UserCreate) -> UserRead:, to framework jest w stanie wygenerować z tego pełny opis schematu JSON wejścia i wyjścia. To z kolei idealnie pasuje do standardu OpenAPI, którego używa się do opisania REST API w sposób automatyzowalny i zrozumiały dla narzędzi takich jak Swagger UI.

Przygotowanie środowiska – od zera do pierwszego „hello world”

Wymagania wstępne i wybór narzędzi

Do pracy z FastAPI przydają się trzy elementy:

• Podstawowa znajomość Pythona – funkcje, moduły, proste klasy, typy w adnotacjach.
• pip – menedżer pakietów Pythona, zazwyczaj instalowany razem z Pythonem.
• Wirtualne środowisko – izolacja zależności projektu. Najprostsza opcja to venv, ale wiele osób korzysta też z poetry lub conda.

Przykładowe utworzenie środowiska z venv:

python -m venv venv
# Linux / macOS
source venv/bin/activate
# Windows (PowerShell)
venvScriptsActivate.ps1

Po aktywacji środowiska każda instalacja pakietu przez pip wyląduje w katalogu projektu, a nie w systemowym Pythonie. Chroni to przed konfliktem wersji między różnymi projektami.

Instalacja FastAPI i Uvicorn krok po kroku

FastAPI jest tylko frameworkiem – aby serwować go przez HTTP, potrzebny jest serwer ASGI. Najczęściej używa się Uvicorn. Instalacja wygląda następująco:

pip install "fastapi[all]" uvicorn

Opcja "fastapi[all]" instaluje także kilka przydatnych dodatków (np. do renderowania dokumentacji). W prostych projektach wystarczy również:

pip install fastapi uvicorn

Wspomniane wcześniej ASGI to nowszy standard interfejsu pomiędzy serwerem a aplikacją w Pythonie, który obsługuje asynchroniczność. Starszy WSGI (na którym działają np. Flask czy Django w klasycznym trybie) zakłada synchroniczny model obsługi żądań. FastAPI działa na ASGI, więc potrafi wykorzystać async def w pełni.

Minimalna struktura projektu i pierwszy endpoint

Nawet bardzo mały projekt dobrze jest uporządkować. Na początek wystarczy jedna struktura katalogów:

projekt-fastapi/
├── venv/
└── main.py

Zawartość pliku main.py z pierwszym działającym „hello world”:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello World from FastAPI"}

W tym fragmencie dzieje się kilka rzeczy:

• tworzona jest instancja aplikacji app = FastAPI(),
• dekorator @app.get("/") oznacza funkcję jako endpoint HTTP GET pod ścieżką /,
• zwracany jest słownik Pythona, który FastAPI automatycznie konwertuje do JSON.

Uruchamianie serwera deweloperskiego z auto-reload

Aplikację uruchamia się za pomocą Uvicorna. W katalogu z main.py wykonaj:

uvicorn main:app --reload

Wyjaśnienie parametrów:

• main:app – main to nazwa modułu (plik main.py), app to obiekt aplikacji FastAPI,
• --reload – serwer automatycznie przeładuje się po zmianie kodu, co mocno przyspiesza pracę.

Dodatkowe przydatne opcje:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

--host 0.0.0.0 sprawi, że serwer będzie dostępny z innych urządzeń w sieci lokalnej (np. z telefonu), a --port pozwala zmienić domyślny port 8000 na inny.

Realny przykład: mała wewnętrzna usługa w godzinę

Typowa sytuacja biurowa: zespół potrzebuje prostego API, które na podstawie identyfikatora użytkownika zwróci jego dane z bazy lub z pliku CSV. Często taki problem rozwiązuje się małym skryptem uruchamianym lokalnie. FastAPI pozwala zamiast tego zbudować mini-serwis HTTP, do którego może podłączyć się kilka narzędzi w firmie. Parę prostych endpointów, kilka modeli Pydantic, start Uvicorna i po godzinie cała firma może odpytywać dane jednym spójnym sposobem, zamiast kopiować skrypty między komputerami.

Podstawy tworzenia endpointów – od prostych ścieżek do parametrów zapytania

Definiowanie ścieżek z dekoratorami HTTP

FastAPI używa dekoratorów odpowiadających metodom HTTP:

• @app.get() – odczyt danych,
• @app.post() – tworzenie nowych zasobów,
• @app.put() – pełna aktualizacja zasobu,
• @app.patch() – częściowa aktualizacja,
• @app.delete() – usuwanie.

Przykład prostych endpointów:

from fastapi import FastAPI

app = FastAPI()

@app.get("/status")
def get_status():
    return {"status": "ok"}

@app.post("/items")
def create_item():
    return {"message": "Item created"}

@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    return {"deleted_id": item_id}

Dzięki dekoratorom kod jest czytelny: od razu widać, jaka metoda i jaka ścieżka odpowiada za daną funkcję biznesową.

Ścieżki dynamiczne i typowanie parametrów

Dynamiczne fragmenty ścieżki zapisuje się w nawiasach klamrowych, np. /users/{user_id}. Nazwa w klamrach musi odpowiadać nazwie parametru funkcji. Typ tego parametru określasz za pomocą adnotacji:

@app.get("/users/{user_id}")
def read_user(user_id: int):
    return {"user_id": user_id}

FastAPI:

• odczyta fragment ścieżki,
• spróbuje skonwertować go do typu int,
• w przypadku niepowodzenia (np. /users/abc) zwróci błąd 422 z opisem walidacji.

Jeśli użyjesz user_id: str, parametr zostanie potraktowany jako tekst. Typowanie parametrów ścieżki jest jednym z miejsc, gdzie FastAPI od razu pokazuje swoją siłę: błędy danych wejściowych widać jeszcze zanim kod logiki biznesowej się wykona.

Parametry query – filtrowanie i paginacja

Parametry query to wszystko, co pojawia się w adresie po znaku zapytania, na przykład: /items?skip=10&limit=20. FastAPI automatycznie rozpoznaje parametry funkcji, które nie są częścią ścieżki, jako parametry query:

@app.get("/items")
def list_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

W tym przykładzie:

• skip i limit są opcjonalne – jeśli klient ich nie poda, zostaną użyte wartości domyślne 0 i 10,
• FastAPI zadba o konwersję typów i walidację (np. wymusi konwersję na int).

Można też użyć typów opcjonalnych (np. Optional[str]), gdy parametr ma być naprawdę nieobowiązkowy:

from typing import Optional

@app.get("/search")
def search_items(q: Optional[str] = None):
    return {"query": q}

Różnica między parametrem ścieżki, query i body

Dobranie właściwego miejsca na dane jest kluczowe dla czytelności API:

• parametr ścieżki – identyfikuje konkretny zasób: /users/42, /orders/2023-01-01,
• parametr query – modyfikuje sposób pobierania lub filtrowania danych: /users?active=true&page=2,

Przesyłanie danych w body żądania – kiedy POST, a kiedy PUT

Gdy klient wysyła bardziej złożone dane (np. cały obiekt użytkownika), trafiają one do body żądania. W FastAPI najczyściej używa się do tego metod POST (tworzenie) i PUT/PATCH (aktualizacja). Na prostym typie widać to tak:

from fastapi import FastAPI

app = FastAPI()

@app.post("/echo")
def echo_message(message: str):
    return {"received": message}

FastAPI domyśli się, że message ma pochodzić z body, a nie z query, jeżeli:

• nie jest częścią ścieżki,
• nie ma wartości domyślnej,
• jego typ nie jest „prostym” parametrem query w danym kontekście (tu wchodzi w grę system heurystyk FastAPI).

Ten prosty przykład szybko się jednak komplikuje, gdy obiekt ma więcej pól. Wtedy na scenę wchodzi Pydantic.

Walidacja danych z Pydantic – serce wygody w FastAPI

Modele danych jako centralne źródło prawdy

Pydantic to biblioteka, która łączy dwie rzeczy: typy Pythona i walidację danych. Definiujesz klasę opisującą strukturę danych, a Pydantic dba o konwersję, sprawdzanie typów i generowanie czytelnych komunikatów o błędach. Dla FastAPI to idealny partner – modele Pydantic stają się kontraktem pomiędzy klientem a serwerem.

from pydantic import BaseModel

class UserCreate(BaseModel):
    email: str
    full_name: str
    age: int

Taki model można od razu wykorzystać w endpointzie:

from fastapi import FastAPI

app = FastAPI()

class UserCreate(BaseModel):
    email: str
    full_name: str
    age: int

@app.post("/users")
def create_user(user: UserCreate):
    return user

FastAPI:

• odczyta JSON z body,
• spróbuje zmapować go na UserCreate,
• w przypadku błędu (np. age jako tekst) zwróci odpowiedź 422 z opisem, które pola są niepoprawne.

Walidacja typów i konwersje „z automatu”

Modele Pydantic nie tylko sprawdzają, czy typ jest poprawny, ale też próbują sensownie konwertować dane. Przykład pokazujący konwersję stringa na liczbę:

class Product(BaseModel):
    name: str
    price: float

@app.post("/products")
def create_product(product: Product):
    return product

Jeżeli klient wyśle:

{
  "name": "Kabel HDMI",
  "price": "12.99"
}

Pydantic spróbuje przekonwertować "12.99" na float. Jeśli się nie uda (np. "dwanaście"), błąd będzie dokładnie opisany w odpowiedzi.

Ograniczenia pól – długość, zakresy, wzorce

Proste typy to dopiero początek. Pydantic pozwala definiować ograniczenia, dzięki czemu walidacja „przy wejściu” bywa bogatsza niż sprawdzanie danych w samej logice biznesowej.

from pydantic import BaseModel, Field, EmailStr, conint

class UserCreate(BaseModel):
    email: EmailStr
    full_name: str = Field(..., min_length=3, max_length=100)
    age: conint(ge=0, le=120)  # wiek 0–120

Dzieje się tu kilka rzeczy:

• EmailStr sprawdza, czy ciąg znaków wygląda jak adres e‑mail,
• Field(..., ...) pozwala ustawić długość i opis pola,
• conint ogranicza zakres liczby całkowitej.

Jeżeli klient wyśle zbyt krótkie full_name albo wiek wykraczający poza zakres, API od razu odpowie stosownym komunikatem, zamiast przepuszczać niepoprawne dane głębiej w system.

Modele odpowiedzi – separacja inputu od outputu

Dane wejściowe rzadko są identyczne z danymi wyjściowymi. Przykład: podczas tworzenia użytkownika klient nie podaje identyfikatora ani daty utworzenia, ale w odpowiedzi dobrze je zwrócić. W FastAPI robi się to przez modele odpowiedzi (response models).

from datetime import datetime
from pydantic import BaseModel
from fastapi import FastAPI

app = FastAPI()

class UserCreate(BaseModel):
    email: str
    full_name: str

class UserRead(BaseModel):
    id: int
    email: str
    full_name: str
    created_at: datetime

fake_db = []

@app.post("/users", response_model=UserRead)
def create_user(user: UserCreate):
    new_id = len(fake_db) + 1
    user_data = UserRead(
        id=new_id,
        email=user.email,
        full_name=user.full_name,
        created_at=datetime.utcnow(),
    )
    fake_db.append(user_data)
    return user_data

Parametr response_model ma kilka zalet naraz:

• odcina to, czego nie chcesz wystawiać światu (np. hashe haseł, wewnętrzne klucze),
• gwarantuje, że odpowiedź ma przewidywalną strukturę,
• generuje automatyczny opis w dokumentacji OpenAPI.

Modele zagnieżdżone – struktury złożonej odpowiedzi

Rzeczywiste API rzadko zwraca pojedynczy płaski obiekt. Częściej jest to lista rekordów, paginacja, metadane. Modele Pydantic można dowolnie zagnieżdżać.

from typing import List

class Item(BaseModel):
    id: int
    name: str

class PaginatedItems(BaseModel):
    items: List[Item]
    total: int
    page: int
    size: int

@app.get("/items", response_model=PaginatedItems)
def list_items(page: int = 1, size: int = 10):
    # uproszczony przykład, bez prawdziwej bazy
    data = [
        Item(id=1, name="Item 1"),
        Item(id=2, name="Item 2"),
    ]
    return PaginatedItems(
        items=data,
        total=len(data),
        page=page,
        size=size,
    )

Dzięki zagnieżdżonym modelom można precyzyjnie opisać nawet bardzo złożone odpowiedzi, a FastAPI automatycznie uwzględni je w dokumentacji.

Konfiguracja modeli – aliasy, przykłady, ukryte pola

Gdy API musi współpracować z istniejącym systemem, nazwy pól nie zawsze można dobrać idealnie. Pydantic pozwala użyć tzw. aliasów – klient wysyła inne nazwy niż te, których używasz w kodzie.

class LegacyUser(BaseModel):
    user_id: int = Field(alias="userId")
    full_name: str = Field(alias="fullName")

    class Config:
        allow_population_by_field_name = True

Klient może wysłać JSON:

{
  "userId": 1,
  "fullName": "Jan Kowalski"
}

a w kodzie pracujesz z polami user_id i full_name. To spore ułatwienie przy integracjach z „historycznymi” API.

Dokumentacja „za darmo” – Swagger UI i OpenAPI w praktyce

Automatyczny interfejs Swagger UI

Po uruchomieniu aplikacji FastAPI z przykładowego pliku main.py wystarczy wejść w przeglądarce na adres:

http://127.0.0.1:8000/docs

Wyświetli się Swagger UI – interaktywna dokumentacja, która zna wszystkie endpointy, modele, parametry i kody odpowiedzi. Można tam:

• przeglądać listę ścieżek,
• podglądać struktury JSON wejścia i wyjścia,
• wysyłać testowe żądania bezpośrednio z przeglądarki.

W małym zespole to często zastępuje osobne narzędzia do testowania API (np. Postman) w pierwszej fazie pracy.

OpenAPI – kontrakt API w jednym pliku

Pod spodem dokumentacja w FastAPI opiera się na specyfikacji OpenAPI. To ustandaryzowany opis REST API zapisany w formacie JSON lub YAML. W każdej chwili możesz zobaczyć „surową” specyfikację:

http://127.0.0.1:8000/openapi.json

Taki plik można wykorzystać w wielu narzędziach:

• generatorach klientów (np. generowanie klienta w TypeScript lub Java),
• narzędziach do testów automatycznych,
• systemach gateway/API management.

Opisy endpointów, parametrów i odpowiedzi

Sama lista ścieżek to za mało, by zrozumieć API. FastAPI pozwala łatwo uzupełnić dokumentację o opisy tekstowe. Do inicjalizacji aplikacji można podać kilka podstawowych informacji:

app = FastAPI(
    title="Moje API sklepu",
    description="Proste API do zarządzania produktami i zamówieniami.",
    version="0.1.0",
)

Endpointy z kolei można opisać przez argumenty dekoratora:

@app.get(
    "/items",
    summary="Lista przedmiotów",
    description="Zwraca listę wszystkich przedmiotów z możliwością paginacji."
)
def list_items(skip: int = 0, limit: int = 10):
    ...

Te teksty pojawią się w Swagger UI, ułatwiając nowym osobom zrozumienie, do czego służy dany endpoint.

Dokumentowanie kodów odpowiedzi i przykładów

Dobrze opisane API nie kończy się na „200 OK”. FastAPI umożliwia doprecyzowanie, jakie kody odpowiedzi i struktury JSON mogą się pojawić.

from fastapi import status

@app.post(
    "/users",
    response_model=UserRead,
    status_code=status.HTTP_201_CREATED,
    responses={
        400: {"description": "Nieprawidłowe dane wejściowe"},
        409: {"description": "Użytkownik o takim e-mailu już istnieje"},
    },
)
def create_user(user: UserCreate):
    ...

Można także zdefiniować przykłady (examples) dla modeli Pydantic, które będą widoczne w UI:

class UserCreate(BaseModel):
    email: str
    full_name: str

    class Config:
        schema_extra = {
            "example": {
                "email": "user@example.com",
                "full_name": "Jan Kowalski"
            }
        }

Swagger UI pokaże wtedy gotowy przykład requestu, który można jednym kliknięciem wysłać.

Alternatywny interfejs ReDoc

Oprócz Swagger UI, FastAPI dostarcza też drugi interfejs dokumentacji – ReDoc, bardziej przypominający klasyczną dokumentację techniczną. Jest dostępny pod adresem:

http://127.0.0.1:8000/redoc

ReDoc prezentuje endpointy w postaci rozwijanych sekcji z dokładnym opisem typów, co bywa wygodne przy większych projektach, w których drzewo ścieżek zaczyna być rozbudowane.

Asynchroniczność w FastAPI – kiedy używać async, a kiedy nie

Intuicja: dlaczego async może przyspieszyć API

Głównym celem asynchroniczności w API nie jest „szybsze liczenie”, ale lepsze wykorzystanie czasu, gdy aplikacja czeka – na bazę danych, zewnętrzne API, system plików. Zamiast blokować wątek, można w tym czasie obsłużyć inne żądanie.

W FastAPI różnica sprowadza się do tego, czy endpoint jest zdefiniowany jako:

@app.get("/sync")
def read_sync():
    ...

@app.get("/async")
async def read_async():
    ...

Kiedy wystarczy funkcja synchroniczna

Jeśli endpoint:

• robi proste przeliczenia w pamięci,
• nie wywołuje zewnętrznych usług,
• nie korzysta z async‑owych bibliotek (np. async ORM-u),

wtedy zwykłe def jest zupełnie wystarczające. Uvicorn i tak uruchamia takie funkcje w odpowiednim wątku, a kod bywa prostszy do zrozumienia dla osób przyzwyczajonych do tradycyjnego Pythona.

Kiedy naprawdę zyska się na async

Asynchroniczna funkcja endpointu ma sens, gdy:

• korzysta z async‑owego klienta HTTP (np. httpx.AsyncClient),
• używa async ORM-u lub drivera do bazy (np. SQLAlchemy w trybie async, asyncpg),
• wykonuje wiele zewnętrznych wywołań równolegle.

import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/weather")
async def get_weather(city: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(
            "https://api.example.com/weather",
            params={"city": city},
            timeout=5.0,
        )
    return r.json()

Tutaj await pozwala serwerowi obsługiwać inne żądania, gdy trwa oczekiwanie na odpowiedź z zewnętrznego API.

Nie mieszać stylów „na siłę”

Możliwe jest wywoływanie funkcji synchronicznych z async (i odwrotnie), ale nadmierne mieszanie stylów potrafi wprowadzić chaos. Kilka prostych reguł, które ułatwiają utrzymanie porządku:

• jeśli logika jest głównie I/O (operacje wejścia/wyjścia), preferuj async,
• jeśli logika jest głównie CPU (ciężkie obliczenia) – async nie pomoże, rozważ osobny worker / kolejkę zadań,
• nie „udawaj” async, jeśli wszystkie zależności i tak są synchroniczne.

Ciężkie obliczenia a blokowanie event loopa

Przenoszenie zadań CPU do osobnych wątków i procesów

Jeśli endpoint wykonuje ciężkie obliczenia (np. przetwarzanie obrazu, szyfrowanie dużych plików, algorytmy machine learning), asynchroniczność nie poprawi sytuacji. Kod i tak „mieli” dane na procesorze, blokując event loop, czyli mechanizm odpowiedzialny za obsługę wielu żądań naraz.

Rozsądniejszym rozwiązaniem jest przeniesienie takiej pracy do osobnego wątku lub procesu. Można do tego użyć wbudowanej w Pythona puli wątków:

import time
from concurrent.futures import ThreadPoolExecutor

from fastapi import FastAPI
import anyio

app = FastAPI()
executor = ThreadPoolExecutor(max_workers=4)

def heavy_cpu_task(x: int) -> int:
    # symulacja kosztownego liczenia
    time.sleep(3)
    return x * x

@app.get("/square")
async def square(x: int):
    loop = anyio.lowlevel.current_async_module().to_thread
    result = await loop.run_sync(heavy_cpu_task, x)
    return {"result": result}

Żądanie nadal chwilę trwa, ale event loop pozostaje wolny i może obsługiwać inne zapytania. Przy bardziej rozbudowanej architekturze idzie się o krok dalej i deleguje zadania CPU do systemu kolejek (np. Celery, RQ) oraz osobnych workerów.

Łączenie async z blokującymi bibliotekami

Często trafia się sytuacja pośrednia: endpoint jest async, ale jakieś wywołanie korzysta z blokującej biblioteki (np. stary klient do bazy, SDK zewnętrznej usługi). Zamiast przepisywać od razu pół projektu, można „opakować” blokujący fragment w wątek roboczy.

import anyio
from fastapi import FastAPI

app = FastAPI()

def legacy_blocking_call(user_id: int) -> dict:
    # np. stary klient HTTP lub biblioteka do SOAP
    ...
    return {"id": user_id, "name": "Jan"}

@app.get("/legacy-user")
async def get_legacy_user(user_id: int):
    user = await anyio.to_thread.run_sync(legacy_blocking_call, user_id)
    return user

Taki kompromis pozwala stopniowo przechodzić na async bez wstrzymania całego projektu.

Łączenie FastAPI z bazą danych – prosta warstwa persystencji

Wybór podejścia: ORM, query builder czy „goły” driver

FastAPI nie narzuca konkretnej biblioteki do komunikacji z bazą. Można użyć klasycznego ORM-u (np. SQLAlchemy), lżejszych narzędzi (Tortoise ORM, GINO, Piccolo) albo bezpośrednich driverów (psycopg2, asyncpg).

W prostych projektach wygodnie jest zacząć od „klasycznego” SQLAlchemy w wersji synchronicznej. W bardziej rozbudowanych systemach, gdzie ruch jest większy, często sięga się po asynchroniczny wariant SQLAlchemy lub inne async ORM-y.

Minimalny przykład z SQLAlchemy (tryb synchroniczny)

Najpierw instalacja potrzebnych pakietów:

pip install "sqlalchemy>=1.4" "psycopg2-binary"  # dla PostgreSQL
# lub
pip install "sqlalchemy>=1.4" "aiosqlite"  # dla SQLite w trybie async

Dla przejrzystości przykład oprze się o SQLite w trybie synchronicznym, dobry na start i do testów lokalnych.

# db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Definicja modelu tabeli przypomina klasy Pydantic, choć to zupełnie inne narzędzie:

# models.py
from sqlalchemy import Column, Integer, String
from .db import Base

class ItemDB(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, nullable=True)

Tworzenie tabel (np. przy starcie aplikacji):

# main.py (fragment)
from fastapi import FastAPI
from .db import Base, engine

app = FastAPI()

Base.metadata.create_all(bind=engine)

Zależność „session” – wstrzykiwanie połączenia do endpointów

Zamiast ręcznie otwierać i zamykać połączenie z bazą w każdym endpointzie, wygodniej jest użyć zależności FastAPI. To mechanizm, który „wstrzykuje” obiekt do funkcji, dbając o jego cykl życia.

# deps.py
from .db import SessionLocal
from sqlalchemy.orm import Session

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

Teraz endpoint może dostać sesję do bazy jako argument:

# main.py (fragment)
from typing import List
from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session

from . import models
from .deps import get_db
from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    description: str | None = None

class ItemRead(BaseModel):
    id: int
    name: str
    description: str | None = None

    class Config:
        orm_mode = True

@app.post("/items", response_model=ItemRead, status_code=status.HTTP_201_CREATED)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = models.ItemDB(name=item.name, description=item.description)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items", response_model=List[ItemRead])
def list_items(db: Session = Depends(get_db)):
    return db.query(models.ItemDB).all()

Kluczowy element to orm_mode = True w modelu Pydantic. Dzięki temu można zwrócić obiekt SQLAlchemy, a Pydantic „wyciągnie” z niego atrybuty, zamiast wymagać czystego słownika.

Asynchroniczne SQLAlchemy – kiedy ruch rośnie

W nowszych wersjach SQLAlchemy pojawiło się natywne wsparcie dla async. Pozwala ono budować endpointy, które nie blokują event loopa podczas czekania na bazę. Schemat jest podobny jak w wersji synchronicznej, ale z kilkoma różnicami w konfiguracji.

# db_async.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite+aiosqlite:///./test_async.db"

engine = create_async_engine(DATABASE_URL, echo=False, future=True)

AsyncSessionLocal = sessionmaker(
    bind=engine,
    expire_on_commit=False,
    class_=AsyncSession,
)

Base = declarative_base()

async def get_async_db():
    async with AsyncSessionLocal() as session:
        yield session

Endpointy korzystają teraz z async/await:

# main_async.py (fragment)
from typing import List
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from .db_async import Base, engine, get_async_db
from .models import ItemDB
from pydantic import BaseModel
import asyncio

app = FastAPI()

class ItemRead(BaseModel):
    id: int
    name: str
    description: str | None = None

    class Config:
        orm_mode = True

@app.on_event("startup")
async def on_startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.get("/async-items", response_model=List[ItemRead])
async def async_list_items(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(ItemDB))
    items = result.scalars().all()
    return items

Ten wariant sprawdza się szczególnie tam, gdzie API intensywnie komunikuje się z bazą i zewnętrznymi usługami jednocześnie.

Pydantic + ORM: rozdzielenie modeli domenowych i bazodanowych

Przy dalszym rozwoju projektu wygodne jest utrzymywanie dwóch warstw modeli:

• modele ORM – odpowiadają strukturze tabel w bazie,
• modele Pydantic – opisują dane „na wejściu” i „na wyjściu” z API.

Pozwala to bez bólu zmieniać sposoby przechowywania danych (np. migracja z SQLite na PostgreSQL, zmiana typów kolumn), nie ruszając publicznego kontraktu API.

# schemas.py
from pydantic import BaseModel

class ItemBase(BaseModel):
    name: str
    description: str | None = None

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    description: str | None = None

class ItemRead(ItemBase):
    id: int

    class Config:
        orm_mode = True

Endpointy pracują wtedy głównie na schemas.py, a szczegóły ORM są schowane w osobnej warstwie (np. w modułach crud.py). Dzięki temu kod logiczny jest czytelniejszy, a migracje bazy mniej bolesne.

Prosty moduł CRUD jako pośrednik

Nawet w małych projektach przydaje się cienka warstwa funkcji, które „opakowują” typowe operacje na bazie: create, read, update, delete (CRUD). Endpointy stają się wtedy krótsze, a logikę łatwiej testować osobno.

# crud.py
from sqlalchemy.orm import Session
from . import models, schemas

def create_item(db: Session, item_in: schemas.ItemCreate) -> models.ItemDB:
    db_item = models.ItemDB(**item_in.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def get_item(db: Session, item_id: int) -> models.ItemDB | None:
    return db.query(models.ItemDB).filter(models.ItemDB.id == item_id).first()

def list_items(db: Session, skip: int = 0, limit: int = 100):
    return (
        db.query(models.ItemDB)
        .offset(skip)
        .limit(limit)
        .all()
    )

Endpoint korzysta z tych funkcji tak, jakby korzystał z dowolnego serwisu biznesowego:

# main.py (fragment)
from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session
from . import crud, schemas
from .deps import get_db

@app.get("/items/{item_id}", response_model=schemas.ItemRead)
def read_item(item_id: int, db: Session = Depends(get_db)):
    item = crud.get_item(db, item_id=item_id)
    if not item:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Przedmiot nie istnieje",
        )
    return item

W większych systemach takie moduły rozrastają się do pełnoprawnych warstw usług, ale schemat startowy pozostaje ten sam.

Transakcje, błędy i spójność danych

Baza danych to nie tylko odczyty. Gdy pojawiają się operacje na kilku tabelach naraz, pojawia się temat transakcji. SQLAlchemy domyślnie używa transakcji na poziomie sesji, ale w bardziej złożonych przypadkach przydaje się jawne sterowanie.

from sqlalchemy.orm import Session
from fastapi import HTTPException, status

def transfer_funds(db: Session, from_id: int, to_id: int, amount: int):
    try:
        from_acc = db.query(Account).get(from_id)
        to_acc = db.query(Account).get(to_id)

        if from_acc.balance < amount:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Brak środków",
            )

        from_acc.balance -= amount
        to_acc.balance += amount

        db.commit()
    except:
        db.rollback()
        raise

Taki wzorzec łączy prostotę z bezpieczeństwem danych: jeśli coś się nie uda, zmiany nie trafiają do bazy.

Przykładowa mini-aplikacja: lista zadań (TODO) z FastAPI i SQLite

Aby połączyć ze sobą omówione klocki, warto mieć w głowie prosty, realistyczny scenariusz. Jednym z klasycznych przykładów jest lista zadań (TODO). API ma obsługiwać:

• tworzenie zadania,
• listowanie wszystkich zadań,
• oznaczanie zadania jako ukończone,
• usuwanie zadania.

Model ORM:

# models.py
from sqlalchemy import Column, Integer, String, Boolean
from .db import Base

class TodoDB(Base):
    __tablename__ = "todos"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    completed = Column(Boolean, default=False)

Modele Pydantic:

# schemas.py
from pydantic import BaseModel

class TodoBase(BaseModel):
    title: str

class TodoCreate(TodoBase):
    pass

class TodoRead(TodoBase):
    id: int
    completed: bool

    class Config:
        orm_mode = True

Endpointy:

# main.py (fragment)
from typing import List
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session

from .db import Base, engine
from .deps import get_db
from . import models, schemas

app = FastAPI()
Base.metadata.create_all(bind=engine)

@app.post("/todos", response_model=schemas.TodoRead, status_code=status.HTTP_201_CREATED)
def create_todo(todo: schemas.TodoCreate, db: Session = Depends(get_db)):
    db_todo = models.TodoDB(title=todo.title)
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo)
    return db_todo

@app.get("/todos", response_model=List[schemas.TodoRead])
def list_todos(db: Session = Depends(get_db)):
    return db.query(models.TodoDB).all()

@app.post("/todos/{todo_id}/complete", response_model=schemas.TodoRead)
def complete_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = db.query(models.TodoDB).get(todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Zadanie nie istnieje")
    todo.completed = True
    db.commit()
    db.refresh(todo)
    return todo

@app.delete("/todos/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = db.query(models.TodoDB).get(todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Zadanie nie istnieje")
    db.delete(todo)
    db.commit()
    return

Taki szkielet, mimo że prosty, obejmuje większość typowych elementów backendu: modele Pydantic, modele ORM, zależności, obsługę błędów, statusy HTTP i podstawową pracę z bazą.