SQLAlchemy ORM и миграции

SQLAlchemy ORM и миграции #

1. Что такое ORM-подход в разработке и какие преимущества он предоставляет при работе с данными? #

Что такое ORM-подход #

ORM — Object-Relational Mapping, то есть объектно-реляционное отображение.

Суть подхода: таблицы базы данных представляются в коде как классы, строки таблиц — как объекты, а поля таблиц — как атрибуты объекта.

Например, вместо прямого SQL:

SELECT * FROM users WHERE id = 1;

в ORM можно писать примерно так:

user = User.objects.get(id=1)

В Django модель обычно соответствует таблице базы данных, а атрибуты модели соответствуют полям таблицы. Django поверх этого автоматически предоставляет API для доступа к данным.

Пример идеи ORM #

class User(models.Model):
    username = models.CharField(max_length=100)
    email = models.EmailField()

Эта модель описывает таблицу user или похожую таблицу в БД.

Дальше можно работать не с SQL напрямую, а с Python-объектами:

user = User.objects.create(
    username="alex",
    email="alex@example.com"
)

user.email = "new@example.com"
user.save()

То есть разработчик мыслит не строками таблицы, а объектами приложения.

Что ORM делает под капотом #

ORM преобразует операции с объектами в SQL-запросы:

User.objects.filter(username="alex")

примерно превращается в:

SELECT * FROM users WHERE username = 'alex';

SQLAlchemy прямо описывает себя как SQL toolkit и Object Relational Mapper, который позволяет работать с SQL через более Pythonic-подход.

Основные преимущества ORM #

1. Меньше ручного SQL #

ORM позволяет писать большую часть CRUD-операций через методы и объекты:

User.objects.create(...)
User.objects.get(...)
User.objects.filter(...)
user.save()
user.delete()

Это уменьшает количество шаблонного SQL-кода и делает код приложения ближе к бизнес-логике.

2. Код становится более читаемым #

Сравнение:

SELECT * FROM notifications
WHERE user_id = 10
ORDER BY created_at DESC
LIMIT 20;

ORM-вариант:

Notification.objects.filter(user_id=10).order_by("-created_at")[:20]

В backend-коде такой вариант часто читается проще, потому что он выражает намерение на уровне приложения.

3. Связь между таблицами описывается через модели #

Например:

class Notification(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    text = models.TextField()

После этого можно обращаться к связанным объектам как к объектам Python:

notification.user.username

То есть ORM скрывает часть работы с JOIN, внешними ключами и связями между таблицами.

4. Автоматическая генерация SQL-запросов #

Разработчик описывает, что нужно получить, а ORM сама формирует SQL.

Например:

User.objects.filter(is_active=True)

ORM сама построит корректный SQL-запрос к базе. В Django для этого используется QuerySet API — специальный интерфейс для построения и выполнения запросов к БД.

5. Удобная работа с транзакциями и состоянием объектов #

В ORM обычно есть механизм отслеживания состояния объектов: создан, изменён, удалён, связан с текущей транзакцией и т.д.

В SQLAlchemy, например, Session управляет ORM-объектами как объектами, связанными со строками базы данных в рамках транзакции.

Упрощённо:

user.name = "new name"
session.commit()

ORM понимает, что объект изменился, и формирует нужный UPDATE.

6. Меньше риска SQL-инъекций при обычном использовании #

Когда ORM строит запросы через параметры, она обычно не склеивает SQL вручную строками. Это снижает риск SQL-инъекций по сравнению с небезопасным ручным SQL вроде:

sql = f"SELECT * FROM users WHERE name = '{username}'"

Но это не значит, что ORM полностью защищает от ошибок: при использовании raw SQL всё равно нужно соблюдать правила безопасной параметризации.

7. Удобнее менять структуру приложения #

ORM-модели становятся центральным описанием данных приложения.

Например, в Django модель считается основным источником информации о данных: она содержит поля и поведение данных, которые приложение хранит.

Это удобно, потому что структура данных видна прямо в коде:

class Product(models.Model):
    title = models.CharField(max_length=255)
    price = models.DecimalField(max_digits=10, decimal_places=2)

Где ORM особенно полезна #

ORM хорошо подходит для:

CRUD-приложений
админок
REST API
личных кабинетов
интернет-магазинов
сервисов с обычными связями между таблицами
быстрой разработки backend-логики

Например, в Django/DRF или FastAPI + SQLAlchemy ORM позволяет быстро описать модели, связи, запросы и бизнес-логику.

Недостатки ORM #

ORM — это не замена пониманию SQL.

Основные проблемы:

может генерировать неэффективные запросы
может приводить к N+1 query problem
сложные SQL-запросы иногда проще писать вручную
абстракция может скрывать реальную стоимость операций
нужно понимать индексы, JOIN, транзакции и план выполнения запроса

Например, такой код может выглядеть безобидно:

notifications = Notification.objects.all()

for notification in notifications:
    print(notification.user.username)

Но без оптимизации он может сделать один запрос за уведомлениями и потом отдельные запросы за пользователями. В Django для таких случаев применяют select_related() и prefetch_related().

Кратко #

ORM-подход — это способ работать с базой данных через объекты языка программирования, а не напрямую через SQL.

Главная польза ORM:

меньше ручного SQL
выше читаемость кода
быстрее разработка
удобная работа со связями
автоматическая генерация запросов
упрощённые CRUD-операции
удобное управление состоянием объектов и транзакциями

Но хороший backend-разработчик должен понимать не только ORM, но и SQL, потому что производительность, индексы, транзакции и сложные запросы всё равно остаются на уровне базы данных.

2. 2 Способа работы с БД в SQLAlchemy #

2 способа работы с БД в SQLAlchemy #

В SQLAlchemy есть два основных подхода:

SQLAlchemy
├── Core
└── ORM

1. SQLAlchemy Core #

Это более низкоуровневый способ работы с БД.

Ты работаешь ближе к SQL:

from sqlalchemy import create_engine, text

engine = create_engine("postgresql+psycopg2://user:pass@localhost/db")

with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users WHERE id = :id"), {"id": 1})
    user = result.fetchone()

Core даёт:

больше контроля над SQL
меньше абстракций
удобство для сложных запросов
работу через Table, MetaData, select, insert, update, delete

Пример через SQL Expression Language:

from sqlalchemy import select

stmt = select(users_table).where(users_table.c.id == 1)

with engine.connect() as conn:
    result = conn.execute(stmt)

Core — это полноценный SQL toolkit и нижний слой SQLAlchemy. ORM построен поверх Core.

2. SQLAlchemy ORM #

Это более высокоуровневый способ.

Ты работаешь не напрямую с таблицами, а с Python-классами:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy import create_engine

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str]

engine = create_engine("postgresql+psycopg2://user:pass@localhost/db")

with Session(engine) as session:
    user = session.get(User, 1)

ORM даёт:

работу с объектами вместо строк таблицы
маппинг классов на таблицы
relationship между моделями
unit of work
identity map
автоматическую синхронизацию изменений с БД через session.commit()

Пример изменения объекта:

with Session(engine) as session:
    user = session.get(User, 1)
    user.username = "new_name"

    session.commit()

ORM в SQLAlchemy — это объектно-реляционный маппер, который позволяет строить доменные модели и синхронизировать изменения объектов с текущей транзакцией.

Коротко разница #

Подход	С чем работаешь	Когда использовать
Core	SQL-выражения, таблицы, соединения	когда нужен контроль над SQL
ORM	Python-классы и объекты	когда удобнее работать через модели

Главное #

Core = ближе к SQL
ORM = ближе к объектной модели Python

В обычной backend-разработке чаще используют ORM, но для сложных запросов, оптимизации или низкоуровневого контроля могут подключать Core. SQLAlchemy официально описывается как Python SQL Toolkit и ORM, то есть он изначально поддерживает оба подхода.

3. Каким образом в SQLAlchemy реализуется связывание таблиц через foreign key и relationship? #

В SQLAlchemy связь между таблицами обычно задаётся двумя уровнями:

ForeignKey     — связь на уровне базы данных
relationship   — связь на уровне ORM-объектов Python

ForeignKey говорит базе данных:
posts.user_id ссылается на users.id.

relationship() говорит ORM:
у объекта Post есть связанный объект User, а у User может быть список Post.

В документации SQLAlchemy relationship() описывается как механизм связи между ORM-классами, а join между ними обычно строится на основе foreign key между таблицами.

Пример: User -> Post #

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(100), unique=True)

    posts: Mapped[list["Post"]] = relationship(
        back_populates="user"
    )


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(255))

    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id")
    )

    user: Mapped["User"] = relationship(
        back_populates="posts"
    )

Что здесь происходит #

1. `ForeignKey` #

user_id: Mapped[int] = mapped_column(
    ForeignKey("users.id")
)

Это колонка user_id в таблице posts.

Она хранит id пользователя, которому принадлежит пост.

На уровне SQL это примерно:

user_id INTEGER REFERENCES users(id)

Важно: в ForeignKey("users.id") указывается имя таблицы и колонки в базе данных, а не имя Python-класса. В документации SQLAlchemy отдельно указано, что при Declarative mapping в ForeignKey указывается именно имя целевой таблицы.

2. `relationship()` со стороны `Post` #

user: Mapped["User"] = relationship(
    back_populates="posts"
)

Это уже не колонка в базе данных.

Это ORM-атрибут, который позволяет получить связанного пользователя:

post = session.get(Post, 1)

print(post.user.username)

То есть вместо ручного запроса по user_id можно работать с объектом:

post.user

3. `relationship()` со стороны `User` #

posts: Mapped[list["Post"]] = relationship(
    back_populates="user"
)

Это обратная связь.

Она позволяет получить все посты пользователя:

user = session.get(User, 1)

for post in user.posts:
    print(post.title)

То есть:

Post.user     → один пользователь
User.posts    → список постов

Для чего нужен `back_populates` #

back_populates связывает две стороны отношения между собой:

User.posts  <->  Post.user

SQLAlchemy понимает, что это две стороны одной и той же связи.

Например:

user = User(username="alex")
post = Post(title="Hello")

user.posts.append(post)

После добавления post в user.posts, SQLAlchemy понимает, что post.user — это тот же самый user.

В документации SQLAlchemy back_populates используется для явного связывания двух relationship() между собой.

Как это выглядит логически #

users
-----
id
username

posts
-----
id
title
user_id  -> users.id

На уровне ORM:

User.posts    -> список объектов Post
Post.user     -> объект User
Post.user_id  -> обычное значение внешнего ключа

Разница между `ForeignKey` и `relationship` #

ForeignKey:
- создаёт связь на уровне таблиц
- хранится в базе данных
- участвует в ограничениях целостности
- нужен для JOIN и проверки ссылок

relationship:
- создаёт связь на уровне Python-объектов
- не создаёт отдельную колонку
- позволяет писать post.user и user.posts
- помогает ORM загружать связанные объекты

Иначе говоря:

post.user_id

это значение из колонки БД.

post.user

это связанный ORM-объект.

Пример создания связанных объектов #

user = User(username="alex")

post = Post(
    title="First post",
    user=user
)

session.add(post)
session.commit()

Здесь можно не указывать user_id вручную.

SQLAlchemy сам поймёт связь через объект user и при сохранении выставит внешний ключ.

Пример запроса #

from sqlalchemy import select

stmt = select(Post).where(Post.user_id == 1)

posts = session.scalars(stmt).all()

Через связь можно обращаться к пользователю:

for post in posts:
    print(post.user.username)

Но здесь важно помнить о загрузке связанных данных. Если постов много, обращение к post.user может привести к дополнительным запросам. Для оптимизации используют стратегии загрузки вроде joinedload() или selectinload().

Типы связей #

One-to-many #

Один пользователь — много постов:

User -> Post

User.posts
Post.user

Many-to-one #

Много постов — один пользователь:

Post -> User

Post.user_id
Post.user

Фактически one-to-many и many-to-one — это две стороны одной связи.

Many-to-many #

Например, посты и теги:

Post <-> Tag

Для этого обычно создаётся промежуточная таблица:

post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)

А потом связь задаётся через secondary:

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)

    tags: Mapped[list["Tag"]] = relationship(
        secondary=post_tags,
        back_populates="posts"
    )


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)

    posts: Mapped[list["Post"]] = relationship(
        secondary=post_tags,
        back_populates="tags"
    )

SQLAlchemy описывает такие базовые паттерны связей в разделе ORM relationship patterns.

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

Можно сформулировать так:

В SQLAlchemy связь таблиц задаётся через ForeignKey и relationship.

ForeignKey описывает внешний ключ на уровне схемы базы данных:
например, Post.user_id ссылается на User.id.

relationship используется уже на уровне ORM и позволяет работать
со связанными объектами напрямую: post.user или user.posts.

Для двусторонней связи обычно указывают relationship с обеих сторон
и связывают их через back_populates. ForeignKey отвечает за структуру
и целостность данных, а relationship — за удобную объектную навигацию
между моделями.

Кратко #

ForeignKey — связь в базе данных.
relationship — связь между ORM-объектами.
back_populates — связывает две стороны relationship.

Пример:

user_id = mapped_column(ForeignKey("users.id"))
user = relationship("User", back_populates="posts")

ForeignKey нужен, чтобы база понимала связь таблиц.
relationship нужен, чтобы Python-код мог удобно работать со связанными объектами.

4. Какие риски и особенности нужно учитывать при использовании relationship в SQLAlchemy? #

Главное #

relationship() в SQLAlchemy удобен, но он не просто “ссылка на другую модель”. Это механизм ORM, который может автоматически делать SQL-запросы, синхронизировать объекты, каскадно удалять данные и участвовать в flush. Поэтому основные риски связаны с производительностью, каскадами, жизненным циклом Session и неоднозначными связями.

1. N+1 problem #

Самый частый риск — неявная lazy-загрузка.

Пример:

users = session.scalars(select(User)).all()

for user in users:
    print(user.posts)

На вид это один запрос, но фактически может быть:

1 запрос  — получить всех пользователей
N запросов — отдельно получить posts для каждого user

SQLAlchemy прямо указывает, что lazy loading часто приводит к проблеме N+1: при загрузке N объектов доступ к lazy-атрибутам может вызвать N+1 SELECT-запросов. Для решения обычно используют eager loading: selectinload(), joinedload(), subqueryload() или запрет случайной lazy-загрузки через raiseload().

Правильнее:

from sqlalchemy.orm import selectinload

stmt = (
    select(User)
    .options(selectinload(User.posts))
)

users = session.scalars(stmt).all()

selectinload() часто является удобным вариантом для коллекций: сначала загружаются родительские объекты, затем связанные объекты отдельным SELECT ... WHERE foreign_key IN (...).

2. Неожиданные SQL-запросы при обращении к атрибуту #

relationship может выполнить SQL-запрос в момент обычного доступа к атрибуту:

post.user
user.posts

Это удобно, но опасно: код выглядит как простое обращение к полю, а на деле может сходить в базу. В SQLAlchemy поведение загрузки relationship настраивается через параметр relationship.lazy или через loader options в select().

Для контроля можно использовать:

from sqlalchemy.orm import raiseload

stmt = select(User).options(raiseload(User.posts))

Тогда случайное обращение к user.posts вызовет ошибку, а не скрытый SQL-запрос. Это полезно в местах, где нельзя допускать неявных запросов, например в API-сериализации.

3. Проблемы в async-коде #

В AsyncSession особенно опасна неявная lazy-загрузка:

user = await session.get(User, 1)
print(user.posts)  # может попытаться выполнить IO без await

В async-коде SQLAlchemy рекомендует заранее загружать связи через eager loading или явно загружать relationship через AsyncSession.refresh(obj, ["relation_name"]). В документации также указано, что для asyncio обычно используют expire_on_commit=False, чтобы не провоцировать последующие неявные загрузки после commit.

Пример:

stmt = (
    select(User)
    .options(selectinload(User.posts))
)

result = await session.scalars(stmt)
users = result.all()

Для FastAPI + SQLAlchemy async это особенно важно: лучше заранее определить, какие связи нужны endpoint-у, чем потом случайно получить ошибку из-за lazy loading.

4. `joinedload()` может размножать строки #

joinedload() делает загрузку через JOIN. Для связей one-to-many и many-to-many это может привести к тому, что строки родительской сущности будут повторяться в результате SQL-запроса.

В SQLAlchemy 2.x при использовании joinedload() для коллекций нужно вызывать .unique() у результата, иначе ORM может выдать ошибку. Документация объясняет это тем, что join размножает строки, а Result.unique() явно оставляет уникальные ORM-объекты по primary key.

Пример:

stmt = select(User).options(joinedload(User.posts))

users = session.scalars(stmt).unique().all()

Для коллекций часто практичнее использовать selectinload(), а joinedload() чаще удобен для many-to-one / one-to-one связей.

5. Каскадные удаления могут удалить больше, чем ожидалось #

relationship() может управлять каскадами:

posts = relationship(
    back_populates="user",
    cascade="all, delete-orphan"
)

Это удобно, когда дочерние объекты не должны существовать без родителя. Но ошибка в каскадах может привести к удалению связанных данных.

Особенно опасно ставить cascade="all, delete" с двух сторон many-to-many связи. Документация SQLAlchemy предупреждает: если delete cascade настроен двусторонне, каскад может пройти по всем связанным объектам, загружая коллекции и удаляя всё, что связано.

Опасный вариант:

class User(Base):
    groups = relationship(
        secondary=user_groups,
        back_populates="users",
        cascade="all, delete"
    )


class Group(Base):
    users = relationship(
        secondary=user_groups,
        back_populates="groups",
        cascade="all, delete"
    )

Для many-to-many обычно удаляют строки из association table, но не сами связанные сущности.

6. Нужно различать ORM cascade и database cascade #

Есть два разных уровня:

ORM cascade       — поведение SQLAlchemy Session
ON DELETE CASCADE — поведение самой базы данных

Например:

user_id = mapped_column(
    ForeignKey("users.id", ondelete="CASCADE")
)

и:

posts = relationship(
    cascade="all, delete-orphan",
    passive_deletes=True
)

passive_deletes=True говорит ORM сильнее полагаться на каскадное поведение базы данных, если внешний ключ настроен с ON DELETE CASCADE или похожим правилом. В документации SQLAlchemy указано, что этот параметр обычно предполагает наличие правила ON DELETE CASCADE или ON DELETE SET NULL на стороне БД.

Риск: можно настроить каскад в ORM, но забыть про БД, или наоборот. Тогда поведение при удалении будет отличаться от ожидаемого.

7. Detached object и закрытая Session #

relationship часто требует активной Session, потому что lazy-загрузка выполняется через неё.

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

user = session.get(User, 1)
session.close()

print(user.posts)  # ошибка, если posts не были загружены заранее

SQLAlchemy указывает, что lazy loads выполняются через Session, к которой привязан объект; если объект отсоединён от Session, операция загрузки завершится ошибкой.

Решение: заранее загрузить связи до закрытия Session:

stmt = (
    select(User)
    .options(selectinload(User.posts))
    .where(User.id == 1)
)

user = session.scalar(stmt)

8. Неоднозначные foreign key связи #

Если между двумя таблицами несколько внешних ключей, SQLAlchemy может не понять, по какой колонке строить relationship.

Пример:

class Order(Base):
    buyer_id = mapped_column(ForeignKey("users.id"))
    seller_id = mapped_column(ForeignKey("users.id"))

    buyer = relationship("User")
    seller = relationship("User")

Здесь две связи к users.id, поэтому нужно явно указать foreign_keys:

buyer = relationship(
    "User",
    foreign_keys=[buyer_id]
)

seller = relationship(
    "User",
    foreign_keys=[seller_id]
)

Документация SQLAlchemy отдельно описывает ситуацию с несколькими join paths: когда между таблицами больше одного пути через foreign key, relationship нужно настраивать явно.

9. Циклические связи и неправильный `back_populates` #

Для двусторонних связей лучше явно использовать back_populates:

class User(Base):
    posts = relationship(
        "Post",
        back_populates="user"
    )


class Post(Base):
    user = relationship(
        "User",
        back_populates="posts"
    )

Риск возникает, когда связи описаны несогласованно: разные имена, неправильные стороны, несколько relationship на одни и те же foreign key. Тогда можно получить предупреждения, конфликты синхронизации или неожиданное состояние объектов.

relationship() обычно сам определяет join по foreign key, но при нестандартных условиях, нескольких путях или пересекающихся связях join нужно настраивать явно.

10. Большие коллекции нельзя бездумно загружать в память #

Если у пользователя миллионы событий, такой код опасен:

user.events

Он может попытаться загрузить всю коллекцию в память.

SQLAlchemy отдельно выделяет работу с большими коллекциями: когда связанная коллекция потенциально очень большая, её полная загрузка может быть слишком дорогой по времени, сети и памяти.

В таких случаях лучше не полагаться на обычный relationship-доступ, а делать отдельный запрос с фильтрацией, пагинацией и лимитом:

stmt = (
    select(Event)
    .where(Event.user_id == user_id)
    .order_by(Event.created_at.desc())
    .limit(50)
)

events = session.scalars(stmt).all()

11. Сериализация может случайно загрузить полбазы #

Это частая проблема в API:

return user

Если сериализатор проходит по relationship-атрибутам, он может случайно вызвать lazy loading:

user.posts
post.comments
comment.author
author.profile
...

В итоге простой endpoint может сделать десятки или сотни SQL-запросов.

Практически лучше:

- явно выбирать нужные связи через selectinload / joinedload
- не возвращать ORM-модель напрямую как произвольный объект
- использовать DTO / Pydantic-схемы с контролируемыми полями
- для сложных ответов писать явный select

12. Session нельзя разделять между потоками или concurrent async tasks #

Relationship-загрузка завязана на Session, а Session — stateful-объект. SQLAlchemy указывает, что Session представляет одну логическую транзакцию и не должен разделяться между конкурентными потоками или asyncio-задачами; для AsyncSession действует то же правило.

Плохая идея:

# одна AsyncSession используется одновременно в нескольких задачах
await asyncio.gather(
    load_user(session, 1),
    load_user(session, 2),
)

В web-приложении обычно делают одну Session / AsyncSession на один request.

Хороший ответ на собеседовании #

При использовании relationship в SQLAlchemy нужно учитывать,
что это не просто ссылка между моделями, а механизм ORM-загрузки
и синхронизации объектов.

Главные риски: N+1 problem из-за lazy loading, неожиданные SQL-запросы
при обращении к атрибутам, проблемы в async-коде, если relationship
пытается загрузиться без await, ошибки при закрытой Session,
а также неправильные cascade-настройки, которые могут удалить больше данных,
чем ожидалось.

Также важно явно настраивать relationship, если между таблицами несколько
foreign key путей, и не загружать большие коллекции через обычный доступ
user.items. Для оптимизации обычно применяют selectinload, joinedload,
raiseload, явные запросы с limit/offset и аккуратную настройку cascade,
passive_deletes и back_populates.

Кратко #

Основные риски relationship:

1. N+1 queries
2. скрытые SQL-запросы при доступе к атрибуту
3. проблемы lazy loading в AsyncSession
4. размножение строк при joinedload
5. опасные cascade delete / delete-orphan
6. конфликт ORM cascade и ON DELETE CASCADE в БД
7. DetachedInstanceError после закрытия Session
8. неоднозначные foreign key связи
9. загрузка больших коллекций в память
10. случайная глубокая сериализация связанных объектов

Практическое правило: relationship удобен для навигации между объектами, но в production-коде загрузку связей лучше контролировать явно.

5. В какой момент ORM подгружает связанные модели при работе со связями? #

В SQLAlchemy связанные модели подгружаются не в один фиксированный момент, а в зависимости от стратегии загрузки relationship.

Главные варианты:

lazy loading   — связь подгружается при первом обращении к атрибуту
eager loading  — связь подгружается сразу вместе с основным запросом или сразу после него
no loading     — связь не подгружается автоматически
raise loading  — при попытке ленивой загрузки выбрасывается ошибка

SQLAlchemy в документации разделяет загрузку relationship на lazy loading, eager loading и варианты отключения/запрета загрузки.

1. По умолчанию: при первом обращении к relationship #

Обычно relationship() работает через lazy loading.

Например:

user = session.get(User, 1)

На этом этапе SQLAlchemy загружает только пользователя:

SELECT * FROM users WHERE id = 1;

Но связанные посты ещё не загружены.

Когда происходит обращение:

user.posts

ORM видит, что posts ещё не загружены, и делает отдельный SQL-запрос:

SELECT * FROM posts WHERE user_id = 1;

То есть при lazy loading связанные данные подгружаются в момент первого доступа к атрибуту relationship, если объект всё ещё связан с активной Session. В глоссарии SQLAlchemy указано, что при обращении к незагруженному relationship ORM использует связанную Session и выполняет SELECT.

2. При eager loading: сразу во время основного запроса #

Если заранее указать стратегию загрузки, связанные модели будут загружены сразу.

Например через selectinload():

from sqlalchemy import select
from sqlalchemy.orm import selectinload

stmt = (
    select(User)
    .options(selectinload(User.posts))
)

users = session.scalars(stmt).all()

Тогда SQLAlchemy сделает примерно так:

1 запрос — получить users
2 запрос — получить posts для найденных users через WHERE user_id IN (...)

То есть user.posts уже будут загружены, и отдельного запроса при обращении к ним не будет. selectinload() входит в основные стратегии relationship loading в SQLAlchemy.

3. При `joinedload()`: в одном SQL-запросе через JOIN #

from sqlalchemy.orm import joinedload

stmt = (
    select(User)
    .options(joinedload(User.posts))
)

users = session.scalars(stmt).unique().all()

В этом случае связанные данные подтягиваются через JOIN в рамках основного запроса.

Упрощённо:

SELECT users.*, posts.*
FROM users
LEFT OUTER JOIN posts ON users.id = posts.user_id;

Но для коллекций joinedload() может размножать строки, поэтому в SQLAlchemy 2.x обычно нужен .unique() при получении результата. Документация SQLAlchemy отдельно описывает joinedload() как eager loading стратегию.

4. После `commit` данные могут снова стать “незагруженными” #

По умолчанию после session.commit() SQLAlchemy может пометить данные объекта как expired.

Пример:

user = session.get(User, 1)

session.commit()

user.posts  # может вызвать новый запрос

То есть даже если объект уже был загружен, после завершения транзакции ORM может потребовать повторной загрузки данных при следующем обращении.

В async-коде поэтому часто используют:

async_sessionmaker(
    engine,
    expire_on_commit=False
)

SQLAlchemy в разделе asyncio отдельно рекомендует избегать неявного IO и часто использовать expire_on_commit=False, чтобы не провоцировать ленивые загрузки после commit.

5. В async SQLAlchemy lazy loading особенно опасен #

В синхронном SQLAlchemy такой код может просто сделать дополнительный SQL-запрос:

user.posts

А в async-коде это проблема, потому что обращение к атрибуту не содержит await, но загрузка из базы требует IO.

user = await session.get(User, 1)

user.posts  # опасно, если posts не были загружены заранее

Поэтому в AsyncSession обычно заранее загружают relationship через selectinload():

stmt = (
    select(User)
    .options(selectinload(User.posts))
)

Документация SQLAlchemy по asyncio отдельно описывает предотвращение неявного IO при доступе к ORM-атрибутам.

Основная схема #

Без options:

session.get(User, 1)
    ↓
загружается только User
    ↓
user.posts
    ↓
SQLAlchemy делает отдельный SELECT для posts


С selectinload:

select(User).options(selectinload(User.posts))
    ↓
загружается User
    ↓
отдельным запросом сразу загружаются posts
    ↓
user.posts уже в памяти


С joinedload:

select(User).options(joinedload(User.posts))
    ↓
User и posts загружаются одним запросом через JOIN

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

ORM подгружает связанные модели в зависимости от стратегии загрузки relationship.

По умолчанию используется lazy loading: связанная модель или коллекция
не загружается сразу с основным объектом, а подгружается при первом обращении
к атрибуту, например user.posts или post.user. В этот момент SQLAlchemy
через активную Session выполняет дополнительный SELECT.

Если нужно избежать неявных запросов и N+1 problem, связи загружают заранее
через eager loading: selectinload, joinedload или subqueryload. Тогда связанные
данные будут загружены либо тем же SQL-запросом через JOIN, либо отдельным
запросом сразу после получения основных объектов.

Кратко #

Когда подгружается relationship:

lazy loading:
при первом обращении к атрибуту

selectinload:
сразу после загрузки основных объектов отдельным SELECT

joinedload:
сразу в основном SQL-запросе через JOIN

subqueryload:
сразу после основного запроса через подзапрос

raiseload:
не подгружается, а выбрасывает ошибку при попытке lazy loading

Главное правило: relationship может выглядеть как обычное поле Python-объекта, но фактически его чтение может привести к SQL-запросу.

6. Что представляет собой проблема N+1 запросов при работе с базой данных или ORM? #

Проблема N+1 запросов возникает, когда приложение сначала выполняет один запрос для получения списка объектов, а затем для каждого объекта выполняет отдельный запрос для получения связанных данных. Вместо одного запроса получается N+1 запросов, где N — количество объектов в выборке.

Django

QuerySet — это фундаментальное понятие в Django, которое представляет собой коллекцию объектов из базы данных.

QuerySet — это “ленивый” (lazy) объект, который:

Описывает SQL запрос к базе данных
Не выполняется до момента реального использования данных
Позволяет строить сложные запросы цепочкой методов

N+1 проблема возникает когда:

Сначала выполняется 1 запрос для получения основной коллекции объектов

Затем для каждого объекта выполняется дополнительный запрос для получения связанных данных

class Book(models.Model):
    title = models.CharField(max_length=200)
    authors = models.ManyToManyField(Author)  # ManyToMany

class Author(models.Model):
    name = models.CharField(max_length=100)

    books = Book.objects.all()
    for book in books:
        print(book.title)           # 1 запрос
        print(book.author.name)     # N запросов (по одному на каждую книгу)

Select_related используется для:

Для ForeignKey и OneToOneField
Выполняет JOIN на уровне SQL
Загружает связанные объекты в одном запросе

Синтаксис:

queryset.select_related('related_field')

books = Book.objects.select_related('author').all()

Prefetch_related используется для:

Для ManyToManyField и reverse ForeignKey
Выполняет отдельный запрос для каждого отношения
Оптимизирует загрузку с помощью Python

Синтаксис:

queryset.prefetch_related('related_field')

authors = Author.objects.prefetch_related('book_set').all()

SQLAlchemy ORM-подход (объекты)

Проблемный код (N+1):

authors = session.query(Author).all()  # 1 запрос
for author in authors:
    print(len(author.posts))  # Каждая итерация → новый запрос (N раз)

Решение через joinedload (один запрос с JOIN):

from sqlalchemy.orm import joinedload

authors = session.query(Author).options(
    joinedload(Author.posts)
).all()  # 1 запрос с JOIN

Решение через selectinload (два запроса с IN):

from sqlalchemy.orm import selectinload

authors = session.query(Author).options(
    selectinload(Author.posts)
).all()  # 2 запроса: авторы + посты WHERE id IN (...)

SQLAlchemy Core-подход (таблицы)

Проблемный код (N+1):

authors = conn.execute(select(authors)).fetchall()  # 1 запрос
for author in authors:
    posts = conn.execute(                         # N запросов
        select(posts).where(posts.c.author_id == author.id)
    ).fetchall()
    print(len(posts))

Решение через JOIN (один запрос):

stmt = select(authors, posts).join(
    posts, authors.c.id == posts.c.author_id
)
rows = conn.execute(stmt).fetchall()  # 1 запрос
# Группировка данных в Python

Решение через IN (два запроса):

authors = conn.execute(select(authors)).fetchall()  # 1 запрос
ids = [a.id for a in authors]

posts = conn.execute(
    select(posts).where(posts.c.author_id.in_(ids))
).fetchall()  # 2-й запрос
# Группировка постов по author_id в Python

Сравнение подходов

	Django	SQLAlchemy ORM	SQLAlchemy Core
Foreign key	`select_related`	`joinedload`	`JOIN`
Many-to-many	`prefetch_related`	`selectinload`	`IN (…)`

Статьи на хабре - Как работают select_related и prefetch_related в Django https://habr.com/ru/articles/752574/

Django против N+1 запросов: оптимизация с помощью select_related и prefetch_related https://habr.com/ru/companies/otus/articles/875384/

7. Как синхронизируют миграции базы данных в команде и решают конфликты между ними? #

В команде миграции синхронизируют так же, как код: миграционные файлы коммитят в Git, проходят через PR/MR, применяют в одинаковом порядке на локальной БД, staging и production. В Django миграции прямо считаются “version control system for your database schema”, а файлы миграций должны коммититься вместе с изменениями моделей.

Обычный командный процесс #

Разработчик меняет модель:

class UserFile(models.Model):
    name = models.CharField(max_length=255)

Создаёт миграцию:

python manage.py makemigrations

Проверяет локально:

python manage.py migrate

Коммитит и модель, и миграцию в одном MR/PR:

storage/models.py
storage/migrations/0005_userfile_name.py

Django отдельно указывает, что после проверки миграции её нужно коммитить вместе с изменением модели, чтобы другие разработчики и production получили одну и ту же историю изменений схемы.

Как миграции применяются у других разработчиков #

Другой разработчик делает:

git pull
python manage.py migrate

Фреймворк смотрит, какие миграции уже применены в базе, и применяет только недостающие.

В Django состояние хранится в таблице:

django_migrations

В Alembic / SQLAlchemy обычно используется:

alembic_version

То есть миграции не “синхронизируются вручную” между базами. Синхронизируется код миграций через Git, а каждая база сама применяет недостающие шаги.

Откуда берутся конфликты #

Частая ситуация:

main:
  0001_initial.py
  0002_add_email.py

feature-A:
  0003_add_avatar.py

feature-B:
  0003_add_phone.py

Два разработчика одновременно создали миграцию 0003 в одном приложении.

Важно: в Django номер 0003 сам по себе не является главным идентификатором. Django смотрит на имя миграции и её dependencies. Поэтому две миграции с похожим номером — не всегда проблема, но может появиться конфликт порядка выполнения. Django описывает такую ситуацию: два разработчика могут одновременно создать миграции с одинаковым номером, а порядок определяется зависимостями внутри файлов миграций.

Как решают конфликт в Django #

Вариант 1: auto-merge миграций #

После merge веток Django может сказать, что есть конфликтующие leaf migrations.

Тогда обычно запускают:

python manage.py makemigrations --merge

Django создаёт merge-миграцию примерно такого вида:

class Migration(migrations.Migration):

    dependencies = [
        ("storage", "0003_add_avatar"),
        ("storage", "0003_add_phone"),
    ]

    operations = []

Смысл: новая миграция говорит Django:

эти две ветки истории считаются объединёнными

Обычно operations = [], потому что реальные изменения уже были в двух предыдущих миграциях.

Вариант 2: вручную поправить зависимости #

Иногда лучше сделать линейную историю:

0001_initial
↓
0002_add_email
↓
0003_add_avatar
↓
0004_add_phone

Например, можно переименовать одну миграцию и изменить её dependency:

dependencies = [
    ("storage", "0003_add_avatar"),
]

Так делают, когда одна миграция логически должна идти после другой.

Как решают конфликт в Alembic / SQLAlchemy #

В Alembic аналогичная проблема называется multiple heads: две миграции ссылаются на одного родителя. Официальная документация Alembic описывает это как branch в migration stream.

Проверяют головы:

alembic heads

Если голов несколько, создают merge revision:

alembic merge -m "merge heads" head1 head2

Или:

alembic merge -m "merge heads" heads

Alembic создаёт миграцию, у которой down_revision указывает сразу на несколько предыдущих ревизий:

revision = "53fffde5ad5"
down_revision = ("ae1027a6acf", "27c6a30d7c24")

Документация Alembic прямо описывает merge-файл как миграцию, которая объединяет две или больше head-миграции.

Хорошая практика в команде #

Перед созданием миграции обновить ветку:

git pull origin main

Создать миграцию только после актуализации ветки:

python manage.py makemigrations

Перед MR/PR проверить:

python manage.py migrate
python manage.py makemigrations --check

В CI полезно проверять, что нет забытых миграций:

python manage.py makemigrations --check --dry-run

Также важно не редактировать старые миграции, которые уже могли попасть в main, staging или production. Безопаснее добавлять новую миграцию поверх старой. Исключение — миграция ещё не смержена и используется только в локальной feature-ветке.

Что нельзя делать #

Нельзя удалять уже применённые на production миграции просто потому, что “они мешают”.

Нельзя менять старую миграцию после того, как она ушла другим разработчикам или на сервер.

Нельзя применять миграции вручную SQL-скриптами, а потом не отражать это в миграционной истории проекта.

Нельзя держать изменения моделей без миграций, иначе у других разработчиков код и схема базы разойдутся.

Итоговая схема #

1. Разработчик меняет модели
2. Создаёт миграцию
3. Проверяет migrate локально
4. Коммитит модель + миграцию
5. CI проверяет миграции
6. MR вливается в main
7. Остальные делают git pull + migrate
8. При конфликте создаётся merge migration или правятся dependencies
9. На staging/production применяются те же миграции

Главная идея: миграции — это часть кода проекта. Их синхронизируют через Git, а конфликты решают через зависимости между миграциями: либо merge-миграцией, либо ручным выстраиванием правильного порядка.

8. За счёт чего SQLAlchemy может работать асинхронно? #

SQLAlchemy работает асинхронно не потому, что вся его внутренняя архитектура стала “нативно async”, а за счёт связки:

AsyncEngine / AsyncSession
↓
asyncio event loop
↓
async-compatible DB driver
↓
greenlet-адаптация внутреннего sync-кода SQLAlchemy

Официальная документация SQLAlchemy указывает, что asyncio-поддержка есть для Core и ORM, но используется она через asyncio-compatible dialects, например asyncpg для PostgreSQL.

Главный источник асинхронности — async-драйвер БД #

Сам SQLAlchemy не делает обычный синхронный драйвер асинхронным.

Например:

postgresql+psycopg2://...

это синхронный вариант.

А вот:

postgresql+asyncpg://...

это асинхронный вариант.

Пример:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/db"
)

Здесь важна часть:

+asyncpg

Она говорит SQLAlchemy использовать асинхронный PostgreSQL-драйвер. В документации SQLAlchemy отдельно указано, что async engine должен использовать asyncio-compatible dialect, например asyncpg.

Что делает AsyncEngine #

AsyncEngine — это асинхронная обёртка над обычной концепцией Engine.

Вместо:

from sqlalchemy import create_engine

engine = create_engine(...)

используется:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(...)

Дальше работа идёт через await:

async with engine.connect() as conn:
    result = await conn.execute(select(User))

SQLAlchemy прямо описывает AsyncEngine как asyncio proxy для обычного Engine.

То есть AsyncEngine не просто “новый движок”, а слой, который даёт async API поверх привычных механизмов SQLAlchemy.

Что делает AsyncSession #

В ORM вместо обычного Session используется:

from sqlalchemy.ext.asyncio import AsyncSession

Пример:

async with AsyncSession(engine) as session:
    result = await session.execute(select(User))
    users = result.scalars().all()

Ключевое отличие:

await session.execute(...)

Запрос к базе — это I/O-операция, поэтому она должна быть awaitable.

Зачем нужен greenlet #

Внутри SQLAlchemy исторически много синхронного кода: построение запросов, работа ORM, unit of work, identity map, транзакции, flush и т.д.

Чтобы не переписывать всё ядро SQLAlchemy полностью под async/await, используется greenlet.

Упрощённо:

твой async-код
↓
await session.execute(...)
↓
AsyncSession
↓
greenlet переключает выполнение во внутренний sync-код SQLAlchemy
↓
когда нужен реальный I/O к базе
↓
управление возвращается в asyncio event loop
↓
async-драйвер делает await

В документации SQLAlchemy прямо сказано, что внутренние возможности SQLAlchemy интегрированы с помощью greenlet, чтобы “протолкнуть” await наружу от database driver к пользовательскому async API.

Важно: это не отдельные потоки #

Асинхронный SQLAlchemy не запускает каждый запрос в отдельном thread pool по умолчанию.

Документация SQLAlchemy указывает, что asyncio API интегрируется с пользовательским event loop, обычно стандартным Python asyncio, без дополнительных потоков или event-систем.

То есть схема такая:

не так:
request → thread → blocking SQL query

а так:
request → coroutine → await DB I/O → event loop может выполнять другие задачи

Пока база отвечает на запрос, event loop может переключиться на другие корутины.

Почему обычный psycopg2 не подойдёт #

Если использовать sync-драйвер внутри async-приложения:

postgresql+psycopg2://...

то запрос к базе будет блокирующим.

Даже если вызвать его внутри async def, он всё равно остановит event loop на время выполнения запроса.

Поэтому для настоящей асинхронной работы нужен async driver:

PostgreSQL → asyncpg / psycopg async mode
SQLite     → aiosqlite
MySQL      → asyncmy / aiomysql

SQLAlchemy отдельно указывает, что async mode требует async driver; иначе возникает ошибка AwaitRequired.

Как это выглядит на практике #

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/app_db"
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    expire_on_commit=False
)

async def get_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User))
        return result.scalars().all()

Здесь асинхронность обеспечивают:

create_async_engine(...)
AsyncSession
await session.execute(...)
asyncpg
asyncio event loop

Почему lazy loading может быть проблемой #

В async ORM опасны неявные запросы к базе.

Например:

user = await session.get(User, 1)

print(user.posts)

Если posts не были заранее загружены, ORM может попытаться сделать lazy load при обычном обращении к атрибуту.

Но обращение к атрибуту:

user.posts

нельзя написать как:

await user.posts

Поэтому в async-режиме такие неявные I/O-операции часто приводят к проблемам. SQLAlchemy указывает, что MissingGreenlet часто возникает, когда I/O был выполнен в неожиданном месте, особенно при lazy loading в ORM.

Правильнее заранее грузить связи:

from sqlalchemy.orm import selectinload

stmt = (
    select(User)
    .options(selectinload(User.posts))
)

result = await session.execute(stmt)
users = result.scalars().all()

Итог #

SQLAlchemy может работать асинхронно за счёт трёх основных вещей:

1. Async API:
   AsyncEngine, AsyncConnection, AsyncSession

2. Async DB driver:
   asyncpg, aiosqlite, asyncmy и т.д.

3. greenlet:
   адаптирует внутренний sync-код SQLAlchemy к async/await-модели

Самая важная мысль:

Асинхронность даёт не ORM сама по себе,
а async-драйвер базы + event loop,
а SQLAlchemy предоставляет удобный async-слой поверх своей Core/ORM архитектуры.

9. Как откатить миграции в Django или Alembic #

Django #

В Django откат делается той же командой migrate, только указывается миграция, к которой нужно вернуться. Django умеет применять и отменять миграции в зависимости от текущего состояния базы. Миграции хранятся как история изменений схемы БД.

Допустим, есть миграции:

storage/migrations/
    0001_initial.py
    0002_add_userfile.py
    0003_add_file_size.py

Если сейчас применена 0003, а нужно откатиться до 0002:

python manage.py migrate storage 0002

Django выполнит обратные операции из 0003.

Откат последней миграции в Django #

Сначала посмотреть список миграций:

python manage.py showmigrations storage

Пример:

[X] 0001_initial
[X] 0002_add_userfile
[X] 0003_add_file_size

Откат на одну миграцию назад:

python manage.py migrate storage 0002

После этого:

[X] 0001_initial
[X] 0002_add_userfile
[ ] 0003_add_file_size

Полный откат миграций приложения в Django #

Чтобы откатить все миграции конкретного приложения:

python manage.py migrate storage zero

zero означает состояние “до первой миграции”. Обычно это удалит таблицы/колонки, созданные миграциями этого приложения. Использовать аккуратно, особенно не на production.

Если миграция уже применена, но файл нужно исправить #

Локально, пока миграция не ушла в main и не применялась другими разработчиками:

python manage.py migrate storage 0002
# исправить модель / миграцию
python manage.py makemigrations
python manage.py migrate

Но если миграция уже попала в общую ветку или production, старую миграцию обычно не редактируют. Правильнее создать новую миграцию, которая исправляет состояние схемы.

Откат через `--fake` в Django #

Иногда нужно изменить только запись в истории миграций, не трогая реальную схему БД:

python manage.py migrate storage 0002 --fake

Это опасный режим. Он говорит Django: “считай, что миграция откатилась”, но SQL-операции реально не выполняются. Используют только когда схема БД уже приведена к нужному состоянию вручную.

Alembic #

В Alembic откат выполняется командой:

alembic downgrade <revision>

В Alembic каждая миграция имеет revision и down_revision. Внутри миграционного файла обычно есть две функции: upgrade() и downgrade(). upgrade() применяет изменение, downgrade() откатывает его.

Пример миграции:

revision = "c12a34"
down_revision = "b98f76"

def upgrade():
    op.add_column("users", sa.Column("age", sa.Integer()))

def downgrade():
    op.drop_column("users", "age")

Откат последней миграции в Alembic #

Откат на один шаг назад:

alembic downgrade -1

Это выполнит downgrade() последней применённой миграции.

Откат к конкретной ревизии в Alembic #

Посмотреть текущую ревизию:

alembic current

Посмотреть историю:

alembic history

Откатиться к конкретной ревизии:

alembic downgrade b98f76

Откатиться полностью к начальному состоянию:

alembic downgrade base

base — состояние до первой миграции.

Пример: Alembic откат и исправление локальной миграции #

alembic downgrade -1
# исправить файл миграции
alembic upgrade head

Так можно делать локально, если миграция ещё не опубликована в общей ветке.

Важное отличие Django и Alembic #

В Django обратные операции часто генерируются автоматически:

migrations.AddField(...)

Django понимает, как откатить AddField.

В Alembic откат зависит от функции:

def downgrade():
    ...

Если downgrade() пустой или неправильный, откат либо ничего не сделает, либо сломается. Поэтому в Alembic важно проверять не только upgrade(), но и downgrade().

Что делать, если откат теряет данные #

Например, миграция удаляет колонку:

op.drop_column("users", "old_email")

Откат может технически вернуть колонку:

op.add_column("users", sa.Column("old_email", sa.String()))

Но данные в этой колонке уже потеряны.

То же самое в Django:

migrations.RemoveField(...)

Обратная миграция может восстановить поле, но не восстановит удалённые значения без backup или отдельной data migration.

На production #

На production откат миграции — не просто команда. Обычно порядок такой:

1. Проверить backup базы
2. Понять, есть ли потеря данных
3. Проверить, есть ли reverse-операции
4. Протестировать откат на staging
5. Откатить приложение и миграцию согласованно
6. Проверить состояние схемы и данные

Для Django:

python manage.py migrate app_name previous_migration

Для Alembic:

alembic downgrade previous_revision

Но если уже выкатили код, который зависит от новой схемы, нужно откатывать и код, и базу согласованно.

Итог #

Django:
    python manage.py migrate app_name migration_name
    python manage.py migrate app_name zero

Alembic:
    alembic downgrade -1
    alembic downgrade <revision>
    alembic downgrade base

Главное правило:

Локальную неопубликованную миграцию можно откатить, исправить и применить заново.

Миграцию, которая уже ушла в main/staging/production, лучше не переписывать.
Нужно создать новую миграцию, которая исправит состояние схемы.

10. Какой инструмент используется для миграций вместе с FastAPI + SQLAlchemy? #

Основной инструмент — Alembic #

В связке:

FastAPI + SQLAlchemy

для миграций обычно используют:

Alembic

Alembic — это официальный инструмент миграций из экосистемы SQLAlchemy. В документации он описан как lightweight database migration tool для SQLAlchemy.

Зачем он нужен #

FastAPI сам по себе не управляет схемой базы данных. Он отвечает за HTTP API:

роуты
валидацию
зависимости
ответы
OpenAPI/Swagger

А SQLAlchemy отвечает за работу с базой:

модели
запросы
ORM/Core
сессии
транзакции

Но изменение структуры БД:

создать таблицу
добавить колонку
удалить индекс
изменить constraint
добавить foreign key

обычно выносят в Alembic.

Типичный набор команд #

Инициализация Alembic в проекте:

alembic init alembic

Создание миграции вручную:

alembic revision -m "create users table"

Создание миграции через сравнение моделей SQLAlchemy с текущей БД:

alembic revision --autogenerate -m "add users table"

Alembic поддерживает автогенерацию миграций через параметр --autogenerate, при котором он сравнивает metadata моделей с текущей схемой базы и создаёт кандидат миграции.

Применить все миграции:

alembic upgrade head

Откатить последнюю миграцию:

alembic downgrade -1

Как это выглядит в проекте #

Обычно структура такая:

project/
    app/
        main.py
        models.py
        database.py
    alembic/
        versions/
            20260626_1200_create_users_table.py
        env.py
    alembic.ini

В models.py лежат SQLAlchemy-модели:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True)

В alembic/env.py обычно подключают metadata:

from app.models import Base

target_metadata = Base.metadata

Именно target_metadata нужен Alembic, чтобы --autogenerate понимал, какие таблицы и поля описаны в SQLAlchemy-моделях.

Важно #

Alembic не заменяет SQLAlchemy и не является частью FastAPI.

Правильное разделение такое:

FastAPI     → HTTP API
SQLAlchemy  → ORM / Core / работа с БД
Alembic     → миграции схемы БД

Для async SQLAlchemy #

Даже если приложение использует:

postgresql+asyncpg://...
AsyncSession
create_async_engine

миграции всё равно обычно делают через Alembic.

В env.py можно настроить async-режим, но сам принцип не меняется:

alembic revision --autogenerate -m "add table"
alembic upgrade head

То есть ответ:

FastAPI + SQLAlchemy → Alembic

11. Как подключиться к базе данных без контекстного менеджера with #

Можно подключиться без with, но тогда нужно самостоятельно закрывать соединение / сессию через close() или await close(). Контекстный менеджер просто делает это автоматически. В SQLAlchemy Connection.close() возвращает DBAPI-соединение обратно в пул соединений, поэтому при ручном управлении закрытие обязательно.

1. SQLAlchemy Core: синхронное подключение без `with` #

Обычно пишут так:

with engine.connect() as conn:
    result = conn.execute(text("SELECT 1"))

Без with:

from sqlalchemy import create_engine, text

engine = create_engine("postgresql+psycopg2://user:password@localhost:5432/db")

conn = engine.connect()

try:
    result = conn.execute(text("SELECT 1"))
    print(result.scalar())
finally:
    conn.close()

Смысл:

engine.connect()  → взять соединение из pool
conn.execute(...) → выполнить запрос
conn.close()      → вернуть соединение обратно в pool

Важно: close() здесь обычно не означает физическое закрытие TCP-соединения. В большинстве случаев соединение возвращается в connection pool SQLAlchemy.

2. SQLAlchemy ORM: синхронная Session без `with` #

Обычно:

with Session(engine) as session:
    users = session.execute(select(User)).scalars().all()

Без with:

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

engine = create_engine("postgresql+psycopg2://user:password@localhost:5432/db")

session = Session(engine)

try:
    users = session.execute(select(User)).scalars().all()
finally:
    session.close()

Если были изменения:

session = Session(engine)

try:
    user = User(username="admin")
    session.add(user)

    session.commit()
except Exception:
    session.rollback()
    raise
finally:
    session.close()

Правильная схема:

commit()   → если всё успешно
rollback() → если была ошибка
close()    → в любом случае

Session.close() закрывает транзакционные ресурсы, используемые ORM-сессией.

3. Async SQLAlchemy: AsyncConnection без `async with` #

Обычно:

async with engine.connect() as conn:
    result = await conn.execute(text("SELECT 1"))

Без async with:

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/db"
)

conn = await engine.connect()

try:
    result = await conn.execute(text("SELECT 1"))
    print(result.scalar())
finally:
    await conn.close()

Здесь обязательно:

await conn.close()

Потому что это асинхронный объект.

4. Async SQLAlchemy: AsyncSession без `async with` #

Обычно:

async with AsyncSessionLocal() as session:
    result = await session.execute(select(User))

Без async with:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/db"
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    expire_on_commit=False
)

session = AsyncSessionLocal()

try:
    result = await session.execute(select(User))
    users = result.scalars().all()
finally:
    await session.close()

С изменениями:

session = AsyncSessionLocal()

try:
    user = User(username="admin")
    session.add(user)

    await session.commit()
except Exception:
    await session.rollback()
    raise
finally:
    await session.close()

AsyncSession — это асинхронный proxy поверх обычной Session, поэтому общая логика жизненного цикла такая же: открыть, выполнить работу, commit/rollback, закрыть.

5. В FastAPI без контекстного менеджера #

Например, зависимость для AsyncSession:

async def get_session():
    session = AsyncSessionLocal()

    try:
        yield session
    finally:
        await session.close()

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

from fastapi import Depends

@app.get("/users")
async def get_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    return result.scalars().all()

Здесь нет async with, но есть finally, который гарантирует закрытие сессии после обработки запроса.

6. Почему `with` обычно лучше #

Без with легко забыть закрыть соединение:

conn = engine.connect()
result = conn.execute(text("SELECT 1"))
# conn.close() забыли

Последствия:

соединения не возвращаются в pool
↓
pool может исчерпаться
↓
новые запросы начнут ждать свободное соединение
↓
приложение может зависать или получать timeout

Поэтому без контекстного менеджера почти всегда нужен такой шаблон:

resource = open_resource()

try:
    ...
except Exception:
    ...
finally:
    resource.close()

Для async:

resource = await open_resource()

try:
    ...
except Exception:
    ...
finally:
    await resource.close()

Итог #

Без with подключаться можно:

conn = engine.connect()

try:
    ...
finally:
    conn.close()

Для ORM:

session = Session(engine)

try:
    ...
finally:
    session.close()

Для async:

session = AsyncSessionLocal()

try:
    ...
finally:
    await session.close()

Главное правило:

Без контекстного менеджера ответственность за закрытие соединения полностью на тебе.

12. Что такое и зачем нужны миграции базы данных? #

Что такое миграции базы данных #

Миграции — это способ управлять изменениями структуры базы данных через код.

Например, в проекте появилась новая модель:

class UserFile(models.Model):
    name = models.CharField(max_length=255)
    size = models.IntegerField()

Чтобы в базе появилась соответствующая таблица или новая колонка, создаётся миграция:

0001_initial.py
0002_add_file_size.py
0003_add_user_avatar.py

Миграция описывает, что нужно изменить в БД:

создать таблицу
добавить колонку
удалить колонку
изменить тип поля
создать индекс
добавить foreign key
изменить constraint

В документации Django миграции прямо описываются как система контроля версий для схемы базы данных. Alembic, который обычно используют с SQLAlchemy, описывается как инструмент для создания, управления и выполнения change management scripts для реляционной БД.

Зачем нужны миграции #

Главная задача миграций — чтобы структура базы данных менялась предсказуемо, повторяемо и синхронно с кодом.

Без миграций ситуация была бы такой:

Разработчик изменил модель в коде
↓
В базе таблица осталась старой
↓
Код ожидает колонку, которой нет
↓
Приложение падает

Например, код обращается к полю:

user.avatar_url

А в базе колонки ещё нет:

SELECT avatar_url FROM users;

Результат — ошибка уровня БД:

column "avatar_url" does not exist

Миграция решает эту проблему: она фиксирует изменение схемы и позволяет применить его на любой базе.

Простая аналогия #

Git хранит историю изменений кода:

commit 1 → добавили модель User
commit 2 → добавили поле email
commit 3 → добавили индекс

Миграции хранят историю изменений базы:

0001_initial        → создали таблицу users
0002_add_email      → добавили колонку email
0003_add_email_idx  → добавили индекс на email

То есть:

Git       → история изменений кода
Migrations → история изменений схемы БД

Что обычно хранится в миграции #

Django #

class Migration(migrations.Migration):

    dependencies = [
        ("accounts", "0001_initial"),
    ]

    operations = [
        migrations.AddField(
            model_name="user",
            name="avatar_url",
            field=models.CharField(max_length=255, null=True),
        ),
    ]

Здесь миграция говорит:

после accounts.0001_initial
добавить поле avatar_url в модель user

Django хранит список доступных migration operations: AddField, RemoveField, CreateModel, AddIndex, AlterField и другие.

Alembic / SQLAlchemy #

def upgrade():
    op.add_column(
        "users",
        sa.Column("avatar_url", sa.String(length=255), nullable=True),
    )


def downgrade():
    op.drop_column("users", "avatar_url")

Здесь:

upgrade()   → применить изменение
downgrade() → откатить изменение

Alembic используется как миграционный инструмент для SQLAlchemy. ( Alembic)

Для чего они нужны в команде #

Миграции позволяют всем разработчикам иметь одинаковую структуру БД.

Разработчик A сделал изменение:

добавил поле avatar_url
создал миграцию 0004_add_avatar_url.py
закоммитил модель + миграцию

Разработчик B делает:

git pull
python manage.py migrate

И его локальная база приводится к нужному состоянию.

Без миграций каждому пришлось бы вручную писать SQL:

ALTER TABLE users ADD COLUMN avatar_url VARCHAR(255);

Это быстро приводит к ошибкам:

у одного разработчика колонка есть
у другого нет
на staging другое состояние
на production третье состояние

Для чего они нужны на production #

На production миграции позволяют безопасно обновлять схему базы вместе с релизом.

Например, новая версия приложения требует новое поле:

release v2:
    код использует users.avatar_url

Перед запуском нового кода применяют миграцию:

python manage.py migrate

или для Alembic:

alembic upgrade head

После этого база соответствует новой версии приложения.

Что миграции не делают автоматически #

Миграции не гарантируют, что изменение безопасно для данных.

Например:

удалить колонку email
переименовать поле
изменить тип varchar на integer
добавить NOT NULL колонку без default

Такие изменения могут привести к потере данных или ошибкам при применении.

Пример опасной миграции:

def upgrade():
    op.drop_column("users", "email")

Откат может вернуть колонку:

def downgrade():
    op.add_column("users", sa.Column("email", sa.String()))

Но старые значения email уже будут потеряны, если не было backup или отдельной data migration.

Виды миграций #

Schema migration #

Меняет структуру базы:

таблицы
колонки
индексы
foreign key
unique constraints

Пример:

ALTER TABLE users ADD COLUMN avatar_url VARCHAR(255);

Data migration #

Меняет данные:

заполнить новое поле
перенести данные из одной таблицы в другую
нормализовать старые значения
создать начальные записи

Пример:

for user in users:
    user.avatar_url = "default.png"
    user.save()

Итог #

Миграции базы данных нужны, чтобы:

1. Хранить историю изменений схемы БД
2. Синхронизировать базу с кодом приложения
3. Применять изменения одинаково на local/staging/production
4. Работать в команде без ручных SQL-правок
5. Откатывать изменения, если rollback возможен
6. Делать изменения БД частью Git-истории проекта

Главная мысль:

Миграции — это версионирование структуры базы данных.

Код меняется через коммиты.
База меняется через миграции.

13. Чем отличается SQLAlchemy ORM от SQLAlchemy Core? #

SQLAlchemy Core — это более низкоуровневый SQL toolkit: ты работаешь с таблицами, колонками, SQL-выражениями и результатами запросов.

SQLAlchemy ORM — это более высокий уровень: ты работаешь с Python-классами и объектами, которые связаны с таблицами БД.

Документация SQLAlchemy описывает Core как schema-centric подход, а ORM — как domain-centric подход. То есть Core больше думает таблицами и SQL, ORM — объектами предметной области.

SQLAlchemy Core #

В Core ты явно описываешь таблицу:

from sqlalchemy import Table, Column, Integer, String, MetaData

metadata = MetaData()

users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("username", String),
)

И строишь SQL-запрос через expression language:

from sqlalchemy import select

stmt = select(users).where(users.c.username == "admin")

Выполнение:

with engine.connect() as conn:
    result = conn.execute(stmt)
    rows = result.fetchall()

Результат — это строки, похожие на данные из SQL-запроса:

for row in rows:
    print(row.id, row.username)

То есть Core ближе к SQL:

SELECT users.id, users.username
FROM users
WHERE users.username = 'admin';

SQLAlchemy ORM #

В ORM ты описываешь не просто таблицу, а Python-класс, связанный с таблицей:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str]

Запрос:

from sqlalchemy import select

stmt = select(User).where(User.username == "admin")

Выполнение через Session:

with Session(engine) as session:
    result = session.execute(stmt)
    user = result.scalar_one()

Результат — не просто строка таблицы, а объект User:

print(user.id)
print(user.username)

То есть ORM позволяет работать так:

user.username = "new_admin"
session.commit()

А SQLAlchemy сам сформирует нужный UPDATE.

Главное отличие #

Core:
    работаем с таблицами, колонками, SQL-выражениями

ORM:
    работаем с Python-классами, объектами и связями между ними

Пример разницы:

# Core
stmt = select(users).where(users.c.id == 1)

# ORM
stmt = select(User).where(User.id == 1)

В Core:

users      → объект Table
users.c.id → колонка таблицы

В ORM:

User    → Python-класс модели
User.id → ORM-атрибут модели

Разница в уровне абстракции #

Core #

БД → таблицы → колонки → SQL expressions → rows

Core даёт больше прямого контроля над SQL.

Ты явно думаешь:

из какой таблицы выбрать
какие колонки взять
какой join сделать
какой insert/update/delete выполнить

ORM #

БД → таблицы → Python-классы → объекты → связи

ORM скрывает часть SQL-деталей и даёт объектную модель.

Ты думаешь:

создать User
изменить user.username
получить user.posts
удалить объект
зафиксировать session.commit()

Разница в работе с изменениями #

В Core изменение обычно пишется явно:

stmt = (
    users.update()
    .where(users.c.id == 1)
    .values(username="new_admin")
)

with engine.begin() as conn:
    conn.execute(stmt)

В ORM можно изменить объект:

with Session(engine) as session:
    user = session.get(User, 1)

    user.username = "new_admin"

    session.commit()

ORM сам отслеживает, что объект изменился, и при commit() отправляет нужный SQL.

Это возможно за счёт Session, identity map и unit of work. SQLAlchemy описывает ORM как более высокий уровень, где SQL строится автоматически для работы с объектами.

Разница в связях #

В Core связи — это в основном foreign key и join:

stmt = (
    select(users.c.username, posts.c.title)
    .join(posts, users.c.id == posts.c.user_id)
)

В ORM можно описать отношение между моделями:

from sqlalchemy.orm import relationship


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    posts: Mapped[list["Post"]] = relationship(back_populates="user")


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    user: Mapped["User"] = relationship(back_populates="posts")

И работать объектно:

user = session.get(User, 1)

for post in user.posts:
    print(post.title)

Когда использовать Core #

Core удобен, когда нужен более явный контроль над SQL:

сложные аналитические запросы
массовые insert/update/delete
работа ближе к SQL
минимум ORM-магии
высокий контроль над производительностью

Пример:

stmt = (
    select(users.c.id, users.c.username)
    .where(users.c.is_active == True)
    .order_by(users.c.id)
)

Core часто проще читать, когда задача больше похожа на чистый SQL.

Когда использовать ORM #

ORM удобен, когда приложение работает с бизнес-сущностями:

User
Order
Product
Notification
UserFile

И когда нужно:

работать с объектами
использовать связи relationship
переиспользовать модели в бизнес-логике
отслеживать изменения объектов
делать CRUD в прикладном backend-коде

Пример:

user = User(username="admin")
session.add(user)
session.commit()

Для типичного FastAPI/Django-like backend с сущностями чаще используют ORM.

Важный момент: ORM построен поверх Core #

Это не две полностью независимые библиотеки.

Упрощённо:

SQLAlchemy Core
    ↓
SQL Expression Language, Engine, Connection, Table, Column

SQLAlchemy ORM
    ↓
работает поверх Core и добавляет модели, Session, relationship, identity map

В SQLAlchemy 2.x Core и ORM используют общий стиль запросов через select(). Официальный tutorial SQLAlchemy теперь объединяет Core и ORM в единое обучение, потому что эти части тесно связаны.

Сравнение #

Критерий	Core	ORM
Основная идея	Работа с SQL-выражениями	Работа с Python-объектами
Главный объект	`Table`, `Column`, `Connection`	`class User`, `Session`, `relationship`
Уровень	Ниже	Выше
Результат запроса	Rows / mappings	ORM-объекты
Изменения	Явные `insert/update/delete`	Через изменение объектов
Связи	Через `join`	Через `relationship`
Контроль SQL	Больше	Меньше, но удобнее
Типичный use case	Сложный SQL, batch-операции	CRUD, бизнес-логика, доменные модели

Итог #

SQLAlchemy Core:
    Я сам строю SQL-выражения и работаю ближе к базе.

SQLAlchemy ORM:
    Я описываю Python-модели и работаю с объектами,
    а SQLAlchemy сам строит SQL для загрузки и сохранения.

Главная мысль:

Core — это подход “таблицы и SQL”.

ORM — это подход “классы, объекты и связи”.