Engine, Session и база SQLite

get_db: сессия на запрос через yield и finally

Зависимости с return возвращают значение и на этом заканчиваются. Для ресурсов, которые нужно явно освободить, FastAPI поддерживает зависимости с yield: код до yield выполняется до обработчика, код после — после.

get_db выглядит так:

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

Когда FastAPI видит функцию-генератор в Depends, он запускает её до yield, получает выданный объект и передаёт его в обработчик. После того как обработчик завершился и ответ сформирован, FastAPI продолжает выполнение генератора — то есть исполняет блок finally. Это происходит до отправки HTTP-ответа клиенту (поведение FastAPI 0.115.6).

finally нужен, чтобы db.close() выполнился в любом случае — и при нормальном завершении обработчика, и при исключении. Если убрать try/finally и оставить db.close() просто после yield, то при исключении в обработчике генератор прервётся на yield и строка закрытия не выполнится. Соединение, если сессия его уже заняла, окажется не возвращено в пул вовремя.

Каждый входящий запрос получает свою сессию: Depends(get_db) вызывает get_db() заново. Именно поэтому check_same_thread=False не делает общую сессию потокобезопасной — общей сессии просто нет.

Полный пример

Вот два файла нового минимального проекта. models.py остаётся из предыдущего урока.

database.py:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base

DATABASE_URL = "sqlite:///./roomly.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base.metadata.create_all(bind=engine)

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

main.py:

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from database import get_db

app = FastAPI()

@app.get("/db-check")
def db_check(db: Session = Depends(get_db)):
    return {"status": "ok"}

Запустите из каталога с этими файлами:

uvicorn main:app --reload

При старте Python импортирует database, исполнит create_all и создаст файл roomly.db с таблицей rooms. GET-запрос на /db-check вернёт 200 {"status": "ok"}. Обработчик получает объект Session, но не делает никаких SQL-запросов — маршрут подтверждает только то, что сессия создана и передана без ошибок. Реальные операции с таблицей появятся в следующем уроке.

Обратите внимание: SessionLocal определён до def get_db(), но это не требование к порядку. Python читает тело функции при вызове, а не при объявлении. Имя SessionLocal должно быть разрешено к моменту, когда генератор доходит до строки db = SessionLocal() — то есть когда FastAPI его запускает, а не когда файл импортируется.