Это моя вторая статья. После первой я посмотрел на статистику и понял простую вещь: начинать издалека и пытаться рассказать сразу обо всём — плохая идея. Поэтому сейчас будет одна задача, один странный баг и одна строка, которая его исправила.

Я добавлял в админку сайта портфолио. У проекта есть заголовок, описание и несколько изображений. После загрузки файла API должен вернуть обновлённый проект, чтобы интерфейс сразу показал новую картинку.

Файл сохранялся. Строка появлялась в базе. Повторный GET тоже находил изображение. Но ответ самого запроса загрузки выглядел так:

{
  "id": 42,
  "title": "Новый кейс",
  "cover_url": null,
  "images": []
}

Сначала я решил, что ошибся где-то между flush() и commit(). Оказалось, что база здесь вообще ни при чём. Устаревшим был Python-объект внутри SQLAlchemy-сессии.

Как был устроен код

Модели в упрощённом виде выглядели так:

class Project(Base):
    __tablename__ = "projects"

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

    images: Mapped[list["ProjectImage"]] = relationship(
        "ProjectImage",
        back_populates="project",
        order_by="ProjectImage.sort_order",
        lazy="selectin",
    )


class ProjectImage(Base):
    __tablename__ = "project_images"

    id: Mapped[int] = mapped_column(primary_key=True)
    project_id: Mapped[int] = mapped_column(
        ForeignKey("projects.id", ondelete="CASCADE"),
        nullable=False,
    )
    filename: Mapped[str]
    url: Mapped[str]

    project: Mapped["Project"] = relationship(
        "Project",
        back_populates="images",
    )

Обработчик сначала загружал проект, затем сохранял файл и создавал строку ProjectImage:

async def add_images(self, project_id: int, files: list[UploadFile]):
    project = await self.repository.get_by_id(project_id)

    for upload in files:
        filename = await save_file(upload)

        await self.repository.add_image(
            project_id=project.id,
            filename=filename,
            url=f"/uploads/projects/{project.id}/{filename}",
        )

    return self._to_admin(project)

А репозиторий создавал дочерний объект через внешний ключ:

async def add_image(
    self,
    project_id: int,
    filename: str,
    url: str,
) -> ProjectImage:
    image = ProjectImage(
        project_id=project_id,
        filename=filename,
        url=url,
    )

    self.session.add(image)
    await self.session.flush()
    await self.session.refresh(image)
    return image

На первый взгляд всё логично. flush() отправляет INSERT, refresh(image) перечитывает созданное изображение, у него появляется id. Но в ответ сериализуется не image, а загруженный ранее project.

Что именно протухло

AsyncSession — не просто тонкая обёртка над соединением с базой. Она хранит загруженные ORM-объекты и их состояние. Для одной строки таблицы в рамках сессии обычно существует один Python-объект. Это и есть identity map.

Когда я получил Project, стратегия lazy="selectin" отдельным запросом загрузила и его images. На тот момент изображений ещё не было, поэтому в памяти оказалось вполне корректное значение:

project.images == []

Затем я создал ProjectImage, передав только project_id. SQLAlchemy выполнил INSERT, но из этого не следует, что ORM должна найти уже загруженный объект Project и сама пересобрать его коллекцию images.

Внутри одной транзакции получилось два одновременно правдивых состояния:

• в таблице project_images строка уже есть;

• у Python-объекта project всё ещё загружена старая коллекция [].

Здесь меня подвело слово «relationship». Оно похоже на живую связь, которая всегда отражает базу. На практике это обычный атрибут ORM-объекта со своим состоянием загрузки.

flush() синхронизирует накопленные изменения с базой, но не означает «перечитай все связанные коллекции». refresh(image) тоже обновляет только image. Родительский project от этого свежим не становится.

Отдельно ситуацию делала заметнее настройка сессии:

async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Она удобна в async-приложении: после коммита атрибуты не протухают автоматически и их чтение не пытается внезапно сходить в базу. Но старое состояние тоже остаётся старым, пока приложение явно его не обновит.

Исправление

Мне не нужно было перечитывать весь проект. Достаточно было явно обновить одну коллекцию перед формированием ответа:

async def refresh_images(self, project: Project) -> Project:
    await self.session.refresh(
        project,
        attribute_names=["images"],
    )
    return project

В проекте используется SQLAlchemy 2.0.51, а сама связь настроена через lazy="selectin". Это уточнение важно: поведение загрузчиков и возможности AsyncSession.refresh() в старых версиях могут отличаться.

И вызвать метод после добавления файлов:

for upload in files:
    # сохранение файла
    await self.repository.add_image(
        project.id,
        filename,
        image_url,
    )

await self.repository.refresh_images(project)
return self._to_admin(project)

Теперь SQLAlchemy выполняет запрос для images, заменяет старую коллекцию актуальной, и только после этого Pydantic собирает ответ.

Получается дополнительный SELECT. Для админского запроса загрузки изображений это нормальный обмен: мне важнее вернуть фактическое состояние ресурса, чем сэкономить один запрос и заставить фронтенд отдельно перезагружать карточку.

Почему я выбрал refresh, а не другое решение

Способ не единственный.

Добавлять объект сразу в коллекцию

Можно создавать изображение через родительский объект:

image = ProjectImage(filename=filename, url=url)
project.images.append(image)
await session.flush()

Тогда коллекция меняется в памяти одновременно с созданием строки. Это хороший вариант, если операция действительно принадлежит агрегату Project и коллекция уже загружена.

В моём коде добавление изображения находилось в отдельном методе репозитория и принимало только project_id. Менять его контракт ради неявного побочного эффекта я не захотел. Явный refresh_images() лучше показывает место, где мне требуется актуальное состояние для ответа API.

Сделать повторный SELECT

Можно заново запросить проект с selectinload(Project.images). Но один только повторный запрос не всегда выражает намерение достаточно явно: в сессии уже существует экземпляр с тем же identity. Для принудительного обновления понадобится populate_existing=True либо явное истечение состояния.

Для одной известной коллекции refresh(..., attribute_names=["images"]) короче и понятнее.

Собрать ответ из созданных объектов

Можно сохранить возвращённые ProjectImage в список и вручную дописать их в DTO. Такой вариант убирает дополнительный запрос, но создаёт второе представление состояния. Нужно самостоятельно учитывать сортировку, уже существующие изображения, удаление и вычисление cover_url.

Для горячего участка API это может быть оправданно. Для обычной админки — лишняя сложность.

Проверка, которая теперь ловит регрессию

До этого тест проверял только статус ответа. Этого было недостаточно: сервер честно отвечал 200 OK, хотя тело уже было логически устаревшим.

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

response = await client.post(
    f"/api/v1/admin/projects/{project_id}/images",
    headers=auth,
    files=[
        ("files", ("one.png", io.BytesIO(PNG), "image/png")),
    ],
)

assert response.status_code == 200

images = response.json()["images"]
assert len(images) == 1
assert response.json()["cover_url"] == images[0]["url"]

Для удаления есть симметричная проверка:

response = await client.delete(
    f"/api/v1/admin/projects/{project_id}/images/{image_id}",
    headers=auth,
)

assert response.status_code == 200
assert response.json()["images"] == []

Это важно: тот же эффект работает в обе стороны. После удаления строки загруженная коллекция тоже не обязана самостоятельно стать пустой.

Что я вынес из этого бага

У меня получилось четыре довольно простых правила:

1. flush() гарантирует отправку изменений в базу, но не обновление всех ORM-объектов в памяти.

2. Если relationship уже загружен, относитесь к нему как к снимку состояния, а не как к запросу в реальном времени.

3. После изменения дочерней таблицы через внешний ключ проверьте, должен ли родитель сразу попасть в ответ API.

4. Тестируйте не только код ответа, но и связанные поля JSON, ради которых выполнялась операция.

В документации SQLAlchemy это описано через identity map и состояние объектов, а для явного обновления есть отдельный раздел про refresh и expire. Для AsyncSession загрузка relationship по имени также показана в официальном разделе про asyncio.

Исправление заняло одну строку:

await session.refresh(project, attribute_names=["images"])

Понимание, почему она понадобилась, заняло заметно больше времени. В этом и была вся проблема: база уже знала правду, а объект, который я собирался отдать клиенту, — ещё нет.