SQLAlchemy 2 w FastAPI: modele, relacje i sesja async z SQLite

W tym artykule zbudujesz warstwę danych małej aplikacji z historią rozmów: użytkownicy mają rozmowy, a rozmowy mają wiadomości. Użyjesz SQLAlchemy 2, FastAPI i SQLite z asynchronicznym sterownikiem aiosqlite. Cały kod uruchomiłem lokalnie (Python 3.13, SQLAlchemy 2.1, aiosqlite 0.22, FastAPI 0.159).

Ten temat omawiamy na kursie Python online, w module o bazach danych i sesjach. Zakładam, że wiesz, czym są tabele i klucze obce. Jeśli nie, zacznij od artykułu o relacjach i JOIN w SQL. Pierwszy endpoint w FastAPI opisuję w artykule FastAPI: co to jest i jak stworzyć pierwsze API.

Po co ORM, skoro można pisać SQL?

ORM (Object-Relational Mapper) zamienia wiersze tabeli na obiekty Pythona i odwrotnie. Zamiast składać zapytania w tekście, pracujesz na klasach i atrybutach.

ORM nie zwalnia z myślenia o SQL. Zapytanie, które generuje, nadal trafia do bazy i nadal może być wolne. Warto też wiedzieć, jak działa SQL injection i obrona przed nim, bo text() z doklejonym tekstem użytkownika omija zabezpieczenia ORM.

Jak zainstalować SQLAlchemy 2 i aiosqlite?

pip install "sqlalchemy[asyncio]" aiosqlite fastapi uvicorn

Dodatek [asyncio] instaluje greenlet, z którego korzysta warstwa async. aiosqlite to asynchroniczny sterownik SQLite. SQLite nie wymaga serwera, cała baza to jeden plik, więc nadaje się do nauki i małych aplikacji.

Jak zdefiniować modele w SQLAlchemy 2?

W wersji 2 modele opisujesz adnotacjami typów. Mapped[int] mówi, że kolumna zawiera liczbę, a mapped_column dodaje szczegóły, np. klucz główny albo unikalność. Mapped[str | None] oznacza kolumnę, która może być pusta.

Najpierw plik db.py z klasą bazową, silnikiem i fabryką sesji:

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

engine = create_async_engine("sqlite+aiosqlite:///app.db")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


async def get_session():
    async with SessionLocal() as session:
        yield session

Potem models.py z trzema tabelami:

from datetime import datetime, timezone

from sqlalchemy import DateTime, ForeignKey, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from db import Base


def now() -> datetime:
    return datetime.now(timezone.utc)


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    conversations: Mapped[list["Conversation"]] = relationship(back_populates="user")


class Conversation(Base):
    __tablename__ = "conversations"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    title: Mapped[str] = mapped_column(String(200))
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now)

    user: Mapped["User"] = relationship(back_populates="conversations")
    messages: Mapped[list["Message"]] = relationship(
        back_populates="conversation", cascade="all, delete-orphan"
    )


class Message(Base):
    __tablename__ = "messages"

    id: Mapped[int] = mapped_column(primary_key=True)
    conversation_id: Mapped[int] = mapped_column(ForeignKey("conversations.id"))
    role: Mapped[str] = mapped_column(String(20))
    content: Mapped[str] = mapped_column(Text)

    conversation: Mapped["Conversation"] = relationship(back_populates="messages")

Jak działają relacje jeden do wielu w ORM?

W bazie relację tworzy klucz obcy (ForeignKey("users.id")). W ORM dodajesz do tego relationship, które pozwala przejść od obiektu do obiektu bez pisania JOIN:

• user.conversations zwraca listę rozmów użytkownika,
• conversation.user zwraca właściciela rozmowy,
• back_populates łączy obie strony, więc dopisanie rozmowy do user.conversations ustawia też conversation.user.

cascade="all, delete-orphan" przy wiadomościach oznacza, że usunięcie rozmowy usuwa jej wiadomości. Bez tego baza zostałaby z wiadomościami wskazującymi na nieistniejącą rozmowę.

Dlaczego SQLite domyślnie nie pilnuje kluczy obcych?

To pułapka, która zaskakuje większość osób. Sprawdziłem to na własnej bazie: bez dodatkowego ustawienia SQLite zapisał rozmowę z user_id=999, choć taki użytkownik nie istnieje. Klucze obce są w SQLite domyślnie wyłączone i trzeba je włączyć dla każdego połączenia.

from sqlalchemy import event

engine = create_async_engine("sqlite+aiosqlite:///app.db")


@event.listens_for(engine.sync_engine, "connect")
def enable_foreign_keys(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

Po włączeniu ten sam zapis kończy się błędem IntegrityError: FOREIGN KEY constraint failed. To jest zachowanie, którego oczekujesz od bazy. Dodaj ten fragment do db.py zaraz po utworzeniu silnika.

Jak utworzyć tabele przy starcie aplikacji?

Na etapie nauki wystarczy create_all. Wywołujesz go raz przy starcie aplikacji, w lifespan:

from contextlib import asynccontextmanager

from fastapi import FastAPI

from db import Base, engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(lifespan=lifespan)

create_all tworzy tylko brakujące tabele. Nie zmieni istniejącej tabeli, gdy dodasz kolumnę do modelu. Do tego służą migracje, którym poświęcę osobny artykuł o Alembicie.

Jak działa sesja i dlaczego powinna być jedna na request?

Sesja (AsyncSession) to Twoja „przestrzeń robocza” z bazą. Śledzi obiekty, które wczytałeś lub dodałeś, i zapisuje zmiany przy commit(). Sesja nie jest bezpieczna do współdzielenia między równoległymi requestami. Jeśli dwa requesty używają tej samej sesji, mieszają się ich transakcje i dostajesz trudne do powtórzenia błędy.

Dlatego sesję tworzysz osobno dla każdego requestu i zamykasz po jego zakończeniu. Robi to funkcja get_session z db.py: otwiera sesję w async with, oddaje ją endpointowi przez yield, a po odpowiedzi zamyka. FastAPI wywołuje ją przez Depends, więc nie musisz przekazywać sesji przez kolejne funkcje ręcznie.

W db.py ustawiłem expire_on_commit=False. Domyślnie po commit() SQLAlchemy oznacza wszystkie wczytane obiekty jako przestarzałe i przy następnym odczycie atrybutu sięga po świeże dane. W kodzie synchronicznym to działa niewidocznie. W async taki ukryty odczyt jest niemożliwy. Sprawdziłem to: przy domyślnym ustawieniu user.id po commit() kończy się błędem MissingGreenlet. Wyłączenie wygasania sprawia, że po zapisie możesz czytać atrybuty obiektu bez dodatkowego zapytania.

Jak napisać endpointy, które zapisują i odczytują dane?

Modele odpowiedzi to zwykłe modele Pydantic. FastAPI potrafi zbudować je wprost z obiektów ORM. Więcej o modelach wejścia i wyjścia piszę w artykule o walidacji danych w FastAPI z Pydantic.

from fastapi import Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from db import get_session
from models import Conversation, Message, User


class MessageIn(BaseModel):
    role: str
    content: str


class MessageOut(MessageIn):
    id: int


class ConversationOut(BaseModel):
    id: int
    title: str
    messages: list[MessageOut]


@app.post("/conversations", status_code=201)
async def create_conversation(title: str, session: AsyncSession = Depends(get_session)):
    user = await session.scalar(select(User).where(User.email == "anna@example.com"))
    if user is None:
        user = User(email="anna@example.com")
        session.add(user)
    conversation = Conversation(title=title, user=user)
    session.add(conversation)
    await session.commit()
    return {"id": conversation.id, "title": conversation.title}


@app.post("/conversations/{conversation_id}/messages", status_code=201)
async def add_message(
    conversation_id: int, body: MessageIn, session: AsyncSession = Depends(get_session)
):
    conversation = await session.get(Conversation, conversation_id)
    if conversation is None:
        raise HTTPException(404, "Nie ma takiej rozmowy")
    session.add(Message(conversation_id=conversation.id, **body.model_dump()))
    await session.commit()
    return {"ok": True}


@app.get("/conversations/{conversation_id}", response_model=ConversationOut)
async def get_conversation(conversation_id: int, session: AsyncSession = Depends(get_session)):
    stmt = (
        select(Conversation)
        .where(Conversation.id == conversation_id)
        .options(selectinload(Conversation.messages))
    )
    conversation = await session.scalar(stmt)
    if conversation is None:
        raise HTTPException(404, "Nie ma takiej rozmowy")
    return conversation

Użytkownik „anna@example.com” jest tu na sztywno, bo logowanie i izolację użytkowników zrobimy w ostatnim artykule cyklu. Po wywołaniu endpointów na prawdziwej bazie dostałem:

Trzy rzeczy do zapamiętania z tego kodu:

• select(...) buduje zapytanie, a await session.scalar(...) je wykonuje i zwraca pierwszą wartość (tu: jeden obiekt albo None).
• session.get(Model, id) to skrót do pobrania rekordu po kluczu głównym.
• Zmiany trafiają do bazy dopiero przy await session.commit().

Co oznacza błąd MissingGreenlet i jak go naprawić?

To najczęstszy błąd osób, które przechodzą na SQLAlchemy async. Odtworzyłem go na endpoincie, który pobiera rozmowę przez session.get i zwraca ją z listą wiadomości:

@app.get("/broken/{conversation_id}", response_model=ConversationOut)
async def broken(conversation_id: int, session: AsyncSession = Depends(get_session)):
    return await session.get(Conversation, conversation_id)

Wynik to ResponseValidationError z komunikatem MissingGreenlet: greenlet_spawn has not been called; can't call await_() here. Przyczyna: conversation.messages nie zostało wczytane, więc przy pierwszym dostępie SQLAlchemy próbuje po cichu zapytać bazę. W kodzie synchronicznym by się udało. W async każde zapytanie wymaga await, a odczyt atrybutu nie może czekać, więc dostajesz błąd.

Rozwiązanie to wczytać relację razem z zapytaniem, czyli selectinload, jak w poprawnym endpoincie wyżej. Masz kilka strategii:

Dobry nawyk: w kodzie async zawsze wiedz, które relacje potrzebuje dany endpoint, i ładuj je jawnie.

Czy do SQLite potrzebuję w ogóle wersji async?

Nie zawsze. SQLite działa w tym samym procesie co aplikacja, więc asynchroniczność nie przyspiesza pojedynczego zapytania. Wersja async ma sens, gdy cała aplikacja jest asynchroniczna (jak FastAPI), bo wtedy zapytanie nie blokuje pętli zdarzeń, a ten sam kod zadziała po zmianie bazy na PostgreSQL z asyncpg. Jeśli piszesz mały skrypt albo endpointy def zamiast async def, zwykła synchroniczna sesja jest prostsza. Różnicę między def i async def w FastAPI opisałem w artykule o pierwszym API.

Jedno zastrzeżenie: SQLite zapisuje w jednej transakcji naraz. Przy dużym ruchu zapisów lepszy będzie serwer bazy, taki jak PostgreSQL. Nie testowałem tego tutaj, więc traktuj to jako ogólną zasadę, nie wynik pomiaru.

Jakie błędy mentor poprawia najczęściej?

Najczęstszy błąd przy pierwszej bazie to założenie, że baza pilnuje spójności za Ciebie. SQLite domyślnie nie sprawdza kluczy obcych, więc literówka w user_id przechodzi bez błędu i wychodzi dopiero tydzień później jako „zniknięta” rozmowa. Włącz PRAGMA foreign_keys=ON w pierwszym dniu projektu, nie po pierwszym incydencie.

Co dalej?

Masz działającą warstwę danych, ale zmiana modelu wymaga dziś usunięcia pliku bazy. W kolejnym artykule dodam migracje w Alembicie, żeby zmieniać schemat bez utraty danych. Potem przetestujemy dostęp do bazy w pytest i na końcu zadbamy o to, żeby użytkownik widział tylko własne rozmowy. Wszystkie te zagadnienia omówimy na przyszłym kursie AI developer, który niedługo ogłosimy. Śledź listę naszych ścieżek, żeby nie przegapić startu.