Async Database - szybki dostęp do danych

Witaj! Darwin z asynchronicznymi bazami danych w FastAPI!

Do tej pory używaliśmy in-memory dict. To wygodne na start, ale ma poważną wadę: po każdym restarcie serwera wszystkie obserwacje znikają, jakby ktoś wyrwał kartki z dziennika wyprawy. Teraz podłączymy prawdziwą bazę danych PostgreSQL z async SQLAlchemy!

Analogia Safari: Async database to jak system real-time tracking zwierząt - tysiące obserwacji zapisywanych jednocześnie bez blokowania! Tropiciel, który wysłał meldunek przez radio, nie stoi w miejscu i nie czeka na potwierdzenie z obozu. W tym czasie obserwuje kolejne stado, a odpowiedź odbierze, gdy przyjdzie.

Instalacja

Potrzebujemy dwóch pakietów: samej biblioteki SQLAlchemy z dodatkiem do asynchroniczności oraz sterownika, który rozmawia z PostgreSQL bez blokowania pętli zdarzeń:

1pip install sqlalchemy[asyncio] asyncpg

• SQLAlchemy - ORM
• asyncpg - async PostgreSQL driver

ORM (Object-Relational Mapper) tłumaczy klasy Pythona na tabele, a obiekty na wiersze, więc zamiast pisać SQL ręcznie operujesz zwykłymi obiektami. W powłoce zsh nawiasy kwadratowe mają specjalne znaczenie, dlatego tam wpisz nazwę w cudzysłowie: "sqlalchemy[asyncio]".

Konfiguracja Async SQLAlchemy

Całą konfigurację trzymamy w jednym pliku database.py. Najpierw adres bazy i silnik (engine), czyli obiekt zarządzający połączeniami. Funkcja create_async_engine tworzy jego asynchroniczną wersję:

1# database.py
2from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
3from sqlalchemy.orm import DeclarativeBase
4
5DATABASE_URL = "postgresql+asyncpg://user:password@localhost/safari_db"
6
7# Async engine
8engine = create_async_engine(DATABASE_URL, echo=True)

Przedrostek postgresql+asyncpg w adresie mówi SQLAlchemy, którego sterownika użyć. Opcja echo=True wypisuje w konsoli każde zapytanie SQL - świetne przy nauce, ale na produkcji ją wyłącz. Samo utworzenie silnika jeszcze nie łączy się z bazą, połączenie powstaje przy pierwszym zapytaniu.

Silnik to linia radiowa, a do pracy potrzebujemy sesji - rozmowy z bazą, w której zbieramy zmiany i zatwierdzamy je razem. async_sessionmaker to fabryka takich sesji:

1# Session factory
2AsyncSessionLocal = async_sessionmaker(
3    engine,
4    class_=AsyncSession,
5    expire_on_commit=False
6)

Klasa AsyncSession jest tu domyślna, więc class_ możesz pominąć, choć jawny zapis nie szkodzi. Ważniejsze jest expire_on_commit=False: bez niego po commit() obiekty tracą załadowane wartości, a próba ich odczytu w kodzie async kończy się błędem, bo SQLAlchemy chciałoby niejawnie dociągnąć dane z bazy.

Zostały jeszcze klasa bazowa dla modeli i zależność, która wydaje sesję każdemu żądaniu:

1# Base class
2class Base(DeclarativeBase):
3    pass
4
5# Dependency
6async def get_db():
7    async with AsyncSessionLocal() as session:
8        yield session

DeclarativeBase to punkt, od którego dziedziczą wszystkie tabele. Funkcja get_db używa yield, więc FastAPI przekaże sesję do endpointu, a po wysłaniu odpowiedzi async with sam ją zamknie - nawet gdy endpoint rzuci wyjątek.

Model SQLAlchemy

Model opisuje tabelę. Mapped[int] to adnotacja typu dla kolumny, a mapped_column podaje szczegóły: typ SQL, klucz główny, wartość domyślną:

1# models.py
2from sqlalchemy import Integer, String, Boolean
3from sqlalchemy.orm import Mapped, mapped_column
4from database import Base
5
6class SpeciesModel(Base):
7    __tablename__ = "species"
8
9    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
10    name: Mapped[str] = mapped_column(String(100), nullable=False)
11    scientific_name: Mapped[str] = mapped_column(String(150), nullable=False)
12    population: Mapped[int] = mapped_column(Integer, default=0)
13    endangered: Mapped[bool] = mapped_column(Boolean, default=False)

Nazwa tabeli w bazie pochodzi z __tablename__. Zauważ, że to nie jest model Pydantic z poprzedniej lekcji - SQLAlchemy opisuje, jak dane leżą w bazie, a Pydantic, jak wyglądają w żądaniach i odpowiedziach. W prawdziwym projekcie masz więc oba.

CRUD Operations - Async

CRUD to cztery podstawowe operacje: Create, Read, Update, Delete. Każde zapytanie budujemy funkcją select, a wykonujemy przez await db.execute(...). Zaczynamy od odczytu:

1# crud.py
2from sqlalchemy.ext.asyncio import AsyncSession
3from sqlalchemy import select
4from models import SpeciesModel
5from schemas import SpeciesCreate
6
7async def get_species(db: AsyncSession, species_id: int):
8    result = await db.execute(select(SpeciesModel).where(SpeciesModel.id == species_id))
9    return result.scalar_one_or_none()
10
11async def get_all_species(db: AsyncSession, skip: int = 0, limit: int = 10):
12    result = await db.execute(select(SpeciesModel).offset(skip).limit(limit))
13    return result.scalars().all()

scalar_one_or_none() zwraca jeden obiekt albo None, gdy nic nie pasuje, a scalars().all() - listę obiektów. Parametry offset i limit realizują stronicowanie, więc nie ściągasz od razu całego rezerwatu.

Teraz zapis. Model Pydantic zamieniamy na słownik i rozpakowujemy do konstruktora modelu SQLAlchemy:

1async def create_species(db: AsyncSession, species: SpeciesCreate):
2    db_species = SpeciesModel(**species.model_dump())
3    db.add(db_species)
4    await db.commit()
5    await db.refresh(db_species)
6    return db_species

db.add() tylko dopisuje obiekt do sesji - do bazy trafia on dopiero przy await db.commit(). refresh wczytuje wartości nadane przez bazę, na przykład wygenerowane id. W starszych tutorialach zobaczysz species.dict(), jednak w Pydantic 2 poprawną nazwą jest model_dump().

Usuwanie wykorzystuje funkcję odczytu, którą już mamy:

1async def delete_species(db: AsyncSession, species_id: int):
2    species = await get_species(db, species_id)
3    if species:
4        await db.delete(species)
5        await db.commit()
6    return species

W AsyncSession metoda delete jest korutyną, dlatego potrzebuje await, podobnie jak commit.

FastAPI Endpoints z Async DB

Na koniec łączymy wszystko w endpointach. Depends(get_db) to wstrzykiwanie zależności: FastAPI samo wywoła get_db i poda sesję jako parametr db:

1# main.py
2from fastapi import FastAPI, Depends, HTTPException
3from sqlalchemy.ext.asyncio import AsyncSession
4from database import get_db
5import crud
6import schemas
7
8app = FastAPI()
9
10@app.get("/species", response_model=list[schemas.Species])
11async def list_species(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)):
12    species = await crud.get_all_species(db, skip=skip, limit=limit)
13    return species

Endpoint nie tworzy sesji ani jej nie zamyka - tym zajmuje się zależność. Pozostałe dwa endpointy wyglądają analogicznie:

1@app.get("/species/{species_id}", response_model=schemas.Species)
2async def get_species(species_id: int, db: AsyncSession = Depends(get_db)):
3    species = await crud.get_species(db, species_id)
4    if not species:
5        raise HTTPException(status_code=404, detail="Species not found")
6    return species
7
8@app.post("/species", response_model=schemas.Species)
9async def create_species(species: schemas.SpeciesCreate, db: AsyncSession = Depends(get_db)):
10    return await crud.create_species(db, species)

Aby response_model mógł przyjąć obiekt SQLAlchemy zamiast słownika, model Pydantic musi mieć from_attributes=True, które znasz z lekcji o Pydantic. Brak gatunku zamieniamy na czytelny kod 404.

Zalety async database:

• Tysiące requestów jednocześnie
• Nie blokuje event loop
• Wysoka wydajność

Uczciwie trzeba dodać, że async nie przyspiesza pojedynczego zapytania. Zysk pojawia się wtedy, gdy wiele żądań czeka na bazę jednocześnie - serwer obsługuje inne, zamiast stać bezczynnie. Moja rada: gdy piszesz endpointy async def, używaj też asynchronicznego sterownika, bo zwykłe, blokujące zapytanie wewnątrz async def zatrzymuje całą pętlę zdarzeń.

Następna lekcja: JWT Authentication! Przyda się tu Depends, bo tą samą drogą FastAPI wstrzyknie zalogowanego użytkownika.

Pamiętaj: baza to dziennik wyprawy, sesja to jeden wpis, a commit to podpis, bez którego wpis się nie liczy.