Как подключить базу данных через SQLAlchemy в FastAPI?

Основные компоненты SQLAlchemy

engine управляет подключениями к базе данных, Session выполняет запросы и управляет транзакциями, а ORM-модели описывают таблицы как Python-классы. В FastAPI сессию обычно передают в endpoint через dependency.

Создание engine и сессий

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

DATABASE_URL = 'postgresql+psycopg://user:password@localhost:5432/app_db'

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

Dependency для сессии

Dependency создаёт сессию в начале запроса и гарантированно закрывает её после завершения обработки.

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

Использование в endpoint

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

app = FastAPI()


@app.get('/articles/{article_id}')
def get_article(article_id: int, db: Session = Depends(get_db)):
    article = db.get(Article, article_id)
    if article is None:
        raise HTTPException(status_code=404, detail='Article not found')
    return article

Транзакции

При изменении данных обычно выполняют db.add(), затем db.commit() и db.refresh(), чтобы получить актуальное состояние объекта. При ошибке нужно выполнить rollback() или использовать транзакционный сервисный слой.

db.add(article)
db.commit()
db.refresh(article)

Async-вариант

Если используется асинхронный драйвер и SQLAlchemy async API, применяют create_async_engine и AsyncSession. Важно не смешивать синхронную сессию с await и выбирать подход, который соответствует драйверу и остальному стеку приложения.

Как ответить на собеседовании

Я создаю SQLAlchemy engine и фабрику сессий, затем делаю dependency с yield, которая выдаёт сессию на один запрос и закрывает её в finally. В endpoint получаю Session через Depends(get_db); транзакции и rollback обрабатываю в сервисном слое или явном блоке работы с БД.