Как устроена работа с БД через SQLAlchemy?

Слои SQLAlchemy

SQLAlchemy состоит из двух уровней:

• Core — SQL expression language: Table, MetaData, select(), insert(), update(), delete(), Connection, Engine. Работаете напрямую с таблицами и строками.
• ORM — поверх Core: декларативные модели через DeclarativeBase/Mapped[...], identity map, unit of work, relationships, lazy/eager loading, Session.

Жизненный цикл

1. create_engine(url, pool_size=..., pool_pre_ping=True) — конфигурация пула и диалекта (psycopg, asyncpg, aiomysql, etc.).
2. sessionmaker(bind=engine, expire_on_commit=False) создаёт фабрику сессий.
3. В рамках запроса/юнита работы: with Session() as s: — добавляете/изменяете объекты, s.flush() синхронизирует in-memory изменения с транзакцией, s.commit() фиксирует, s.rollback() откатывает.
4. 2.x-стиль: запросы строятся через select(User).where(...) и исполняются session.execute(stmt).scalars().

Async-вариант (используется в Talento backend)

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

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/jobstream",
    pool_size=10,
    pool_pre_ping=True,
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True, index=True)

async def get_user_by_email(email: str) -> User | None:
    async with SessionLocal() as session:
        stmt = select(User).where(User.email == email)
        result = await session.execute(stmt)
        return result.scalar_one_or_none()

Загрузка связей

• selectinload(User.posts) — отдельный IN-запрос, защищает от N+1.
• joinedload(User.profile) — LEFT OUTER JOIN, хорошо для 1:1.
• raiseload('*') — запретить любые неявные lazy-load в горячем пути.

Подводные камни

• Глобальная единственная Session на приложение — гонки и грязные данные между запросами. Используйте session per request.
• N+1 при сериализации: дергаете obj.posts в цикле, получаете N запросов. Лечится selectinload/joinedload.
• Путать flush и commit: flush отправляет SQL внутри текущей транзакции, commit фиксирует её.
• Использовать ORM-объект после закрытия сессии — DetachedInstanceError на доступе к не-eager полю.
• В async-коде вызывать sync-метод session.query(...) вместо await session.execute(select(...)).
• Игнорировать pool_pre_ping — после рестарта Postgres получаете OperationalError: server closed the connection.
• Делать bulk_insert_mappings и ждать срабатывания ORM-хуков и relationship-каскадов — массовые операции их обходят.

• Путать Engine, Connection и Session.
• Не знать Core vs ORM.
• Игнорировать транзакции.

• Объясняет Core и ORM.
• Понимает роль Session.
• Упоминает transactions и N+1.