1. От SQL-запросов к Python-приложению

В предыдущем уроке мы научились безопасно резервировать задачи в PostgreSQL с помощью транзакций и FOR UPDATE SKIP LOCKED.

Теперь перенесём этот механизм в Python с использованием SQLAlchemy 2.x.

В первом уроке обработчик выглядел следующим образом:

task = get_pending_task()

if task:
    mark_as_processing(task)
    process_task(task)

Теперь вместо двух отдельных операций получения и резервирования задачи мы будем использовать функцию claim_task().

task = claim_task()

if task:
    process_task(task)

claim_task() получает задачу и изменяет её статус на processing в одной транзакции. Функция process_task() по-прежнему отвечает за непосредственное выполнение задачи.

Цель урока: создать Python-приложение, в котором два независимых воркера конкурентно резервируют задачи из общей таблицы PostgreSQL.

Мы проверим:

• Как SQLAlchemy управляет сессиями и транзакциями.
• Как использовать FOR UPDATE SKIP LOCKED из Python.
• Почему резервирование и обновление статуса должны происходить в одной транзакции.
• Почему выполнение задачи лучше отделять от транзакции резервирования.
• Что происходит при аварийном завершении обработки.

2. Подготовка лабораторного проекта

Мы продолжим работать с уже знакомой таблицей backend_training.tasks. Создавать новую таблицу не потребуется.

2.1. Скачайте проект

Скачайте ZIP-архив с материалами третьего урока, прикреплённый к этому шагу, и распакуйте его.

Внутри находится следующая структура:

lesson_03/
├── README_guide.txt
├── 02_reset.sql
├── requirements.txt
├── db.py
├── models.py
├── repository.py
└── worker.py

Мы будем последовательно разбирать эти файлы и использовать их в экспериментах.

2.2. Установите зависимости

Для работы понадобится Python 3.10 или новее, PostgreSQL 16 и два окна терминала.

Перейдите в каталог lesson_03 и выполните:

python3 -m venv .venv

source .venv/bin/activate

python -m pip install -r requirements.txt

Команды активации окружения приведены для macOS/Linux. На Windows они отличаются.

Файл requirements.txt содержит:

SQLAlchemy>=2.0,<2.1
psycopg[binary]>=3.1,<4

SQLAlchemy предоставляет интерфейс для работы с PostgreSQL через Python. psycopg — драйвер соединения с PostgreSQL.

2.3. Настройте подключение

Приложение получает строку подключения из переменной окружения DATABASE_URL.

export DATABASE_URL='postgresql+psycopg://USER:PASSWORD@localhost:5432/backend_training_lab'

Замените USER и PASSWORD своими данными PostgreSQL.

Если пароль содержит специальные символы URL, их необходимо корректно закодировать.

Важно: используйте отдельную учебную базу backend_training_lab. Не запускайте лабораторные SQL-скрипты на рабочих базах данных.

2.4. Восстановите исходные задачи

Убедитесь, что все ранее открытые учебные транзакции завершены, после чего выполните в DBeaver прилагаемый файл 02_reset.sql.

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

Проверьте результат:

SELECT id, type, status
FROM backend_training.tasks
ORDER BY id;

Все три задачи должны иметь статус pending.

3. Подключение к PostgreSQL через SQLAlchemy

Откроем файл db.py.

import os

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


DATABASE_URL = os.environ.get("DATABASE_URL")

if not DATABASE_URL:
    raise RuntimeError("Задайте DATABASE_URL")

url = make_url(DATABASE_URL)

if (
    url.drivername != "postgresql+psycopg"
    or url.database != "backend_training_lab"
):
    raise RuntimeError(
        "Используйте учебную БД backend_training_lab"
    )

engine = create_engine(url, pool_pre_ping=True)

SessionFactory = sessionmaker(bind=engine)

Здесь используются три важных понятия.

Engine

Engine — объект SQLAlchemy, предоставляющий инфраструктуру подключения к базе данных, включая пул соединений.

engine = create_engine(url, pool_pre_ping=True)

Параметр pool_pre_ping=True позволяет проверять соединение при его получении из пула и заменять обнаруженные неработающие соединения. Это не гарантирует защиту от потери соединения во время выполнения транзакции.

SessionFactory

sessionmaker() создаёт фабрику сессий.

SessionFactory = sessionmaker(bind=engine)

Фабрика используется для создания отдельных экземпляров Session.

Session

Session — объект, через который мы выполняем запросы, работаем с ORM-объектами и управляем транзакциями.

Важный принцип: разные одновременно работающие воркеры должны использовать отдельные сессии.

4. ORM-модель задачи

В предыдущем уроке мы непосредственно выполняли SQL-запросы к таблице tasks.

Теперь опишем эту таблицу Python-классом.

Откроем models.py:

from sqlalchemy import Integer, Text
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
)


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"
    __table_args__ = {"schema": "backend_training"}

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

    type: Mapped[str] = mapped_column(Text)

    status: Mapped[str] = mapped_column(Text)

Task — ORM-модель, которая связывает Python-класс с таблицей PostgreSQL.

Mapped и mapped_column() описывают отображение столбцов таблицы на атрибуты Python-объекта.

__table_args__ указывает, что таблица находится в схеме backend_training.

Таблица уже существует в PostgreSQL, поэтому создавать её через SQLAlchemy не нужно.

5. Реализуем безопасное резервирование задачи

Во втором уроке мы использовали следующий SQL:

SELECT id, type, status
FROM backend_training.tasks
WHERE status = 'pending'
ORDER BY id
LIMIT 1
FOR UPDATE SKIP LOCKED;

В SQLAlchemy аналогичную выборку можно описать следующим образом:

stmt = (
    select(Task)
    .where(Task.status == "pending")
    .order_by(Task.id)
    .limit(1)
    .with_for_update(skip_locked=True)
)

Метод with_for_update(skip_locked=True) добавляет к запросу PostgreSQL конструкцию FOR UPDATE SKIP LOCKED.

Это тот же механизм блокировки строк, который мы исследовали в предыдущем уроке.

5.1. Функция claim_task()

Откроем файл repository.py и рассмотрим функцию резервирования:

from sqlalchemy import select

from db import SessionFactory
from models import Task


def claim_task(
    *,
    hold_before_commit: bool = False,
) -> dict | None:

    with SessionFactory.begin() as session:
        stmt = (
            select(Task)
            .where(Task.status == "pending")
            .order_by(Task.id)
            .limit(1)
            .with_for_update(skip_locked=True)
        )

        task = session.scalars(stmt).first()

        if task is None:
            return None

        task.status = "processing"

        result = {
            "id": task.id,
            "type": task.type,
        }

        session.flush()

        if hold_before_commit:
            print(
                f"[ЛАБ] Строка #{task.id} заблокирована, "
                "UPDATE выполнен, транзакция ещё открыта.",
                flush=True,
            )

            input(
                "[ЛАБ] Запустите Worker B; "
                "затем нажмите Enter для COMMIT... "
            )

    return result

Рассмотрим основные действия функции.

Шаг 1. Открываем транзакцию

with SessionFactory.begin() as session:

SQLAlchemy создаёт сессию и управляет транзакцией. При успешном завершении блока выполняется COMMIT, а при исключении — ROLLBACK.

Шаг 2. Выбираем и блокируем задачу

task = session.scalars(stmt).first()

PostgreSQL выбирает первую доступную задачу со статусом pending и блокирует её строку.

Если подходящих незаблокированных задач нет, функция возвращает None.

Это не обязательно означает, что в таблице вообще отсутствуют задачи со статусом pending: некоторые строки могли быть пропущены из-за блокировок.

Шаг 3. Меняем статус

task.status = "processing"

SQLAlchemy отслеживает изменение ORM-объекта.

После этого мы формируем словарь с данными задачи и вызываем:

session.flush()

Что делает flush()?

flush() отправляет накопленные изменения в PostgreSQL в рамках текущей транзакции. В нашем случае выполняется SQL UPDATE.

Однако flush() не выполняет COMMIT.

Шаг 4. Завершаем транзакцию

Обратите внимание: в функции нет явного вызова session.commit().

Он здесь не требуется, потому что SessionFactory.begin() автоматически фиксирует транзакцию при успешном выходе из блока with.

with SessionFactory.begin() as session:
    task.status = "processing"

    session.flush()
    # UPDATE выполнен, но COMMIT ещё не произошёл.

# Успешный выход из блока: COMMIT выполнен.

В нашем примере явный flush() нужен для лабораторного эксперимента: мы хотим отправить UPDATE в PostgreSQL, но оставить транзакцию открытой, чтобы запустить второго воркера.

В обычной реализации явный flush() здесь можно убрать: SQLAlchemy выполнит необходимые изменения перед COMMIT.

with SessionFactory() as session:
    ...
    session.commit()

with SessionFactory.begin() as session:
    ...

5.2. Зачем нужен hold_before_commit?

В функцию добавлен параметр hold_before_commit.

Он используется исключительно в лабораторной работе. Когда параметр равен True, программа ожидает нажатия Enter до завершения транзакции.

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

В production так делать не нужно: искусственное удержание открытой транзакции требуется только для демонстрации механизма.

6. Собираем Python-воркер

Мы уже умеем резервировать задачу. Теперь реализуем её выполнение.

Откроем файл worker.py:

import sys
import time

from repository import claim_task, mark_as_completed


def process_task(task: dict) -> None:
    print(
        f"Началась обработка задачи #{task['id']}",
        flush=True,
    )

    time.sleep(2)

    print(
        f"Задача #{task['id']} обработана",
        flush=True,
    )


def main() -> None:
    if (
        len(sys.argv) not in (2, 3)
        or (
            len(sys.argv) == 3
            and sys.argv[2] != "--hold"
        )
    ):
        raise SystemExit(
            "Использование: python worker.py NAME [--hold]"
        )

    worker_name = sys.argv[1]
    hold = len(sys.argv) == 3

    task = claim_task(hold_before_commit=hold)

    if task is None:
        print(
            f"[{worker_name}] Нет доступных задач",
            flush=True,
        )
        return

    print(
        f"[{worker_name}] Зарезервирована задача #{task['id']}",
        flush=True,
    )

    process_task(task)

    mark_as_completed(task["id"])

    print(
        f"[{worker_name}] Задача завершена",
        flush=True,
    )


if __name__ == "__main__":
    main()

Пока process_task() имитирует выполнение задачи с помощью time.sleep(2).

В реальном сервисе здесь может находиться обращение к внешнему API, создание отчёта или обработка файла.

6.1. Завершаем задачу

После успешного выполнения задачи необходимо изменить её статус с processing на completed.

Для этого в repository.py предусмотрена функция:

def mark_as_completed(task_id: int) -> None:
    with SessionFactory.begin() as session:
        task = session.get(Task, task_id)

        if task is None:
            raise ValueError(
                f"Задача #{task_id} не найдена"
            )

        if task.status != "processing":
            raise ValueError(
                f"Задача #{task_id} имеет статус "
                f"{task.status!r}, ожидался 'processing'"
            )

        task.status = "completed"

Функция mark_as_completed() завершает задачу в отдельной короткой транзакции. Она появилась потому, что сама обработка происходит уже после завершения резервирования.

6.2. Жизненный цикл задачи

pending
   |
   | claim_task()
   | BEGIN
   | SELECT FOR UPDATE SKIP LOCKED
   | UPDATE status = processing
   | COMMIT
   v
processing
   |
   | process_task()
   | Длительная работа вне транзакции
   v
processing
   |
   | mark_as_completed()
   | BEGIN
   | UPDATE status = completed
   | COMMIT
   v
completed

Главный принцип: транзакция резервирования завершается до начала длительной обработки задачи.

Это позволяет не удерживать строковую блокировку PostgreSQL всё время, пока Python выполняет работу.

7. Лабораторная работа: два Python-воркера

Теперь проверим реализованный механизм на практике.

Эксперимент №1. Один воркер

Цель: проверить полный жизненный цикл одной задачи.

Завершите открытые учебные транзакции и выполните 02_reset.sql.

Запустите:

python worker.py A

Перед проверкой таблицы ответьте: какой статус должен получить задача #42 после успешного завершения программы?

Теперь выполните в DBeaver:

SELECT id, type, status
FROM backend_training.tasks
ORDER BY id;

Ожидаемое состояние:

Задача #42 была зарезервирована, обработана и переведена в статус completed.

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

Эксперимент №2. Два конкурентных воркера

Цель: убедиться, что второй воркер пропускает строку, заблокированную первым воркером.

Важно не просто запустить два процесса подряд, а создать ситуацию, когда их транзакции действительно пересекаются.

Завершите предыдущие учебные транзакции и выполните 02_reset.sql.

Откройте два терминала. В обоих должно быть активировано виртуальное окружение и настроена одинаковая переменная DATABASE_URL.

Терминал A

python worker.py A --hold

Дождитесь сообщения:

[ЛАБ] Строка #42 заблокирована,
UPDATE выполнен, транзакция ещё открыта.

Не нажимайте Enter. Worker A сейчас удерживает открытую транзакцию.

Прежде чем продолжить, предположите: какую задачу получит Worker B? Будет ли он ждать завершения Worker A?

Терминал B

python worker.py B

Наблюдайте за выводом программы.

Worker B должен зарезервировать задачу #43, не дожидаясь завершения транзакции Worker A.

Причина — запрос с FOR UPDATE SKIP LOCKED пропускает заблокированную задачу #42.

Пока Worker A продолжает ожидать Enter, выполните обычный SQL-запрос из отдельного подключения DBeaver:

SELECT id, type, status
FROM backend_training.tasks
ORDER BY id;

При стандартном уровне изоляции PostgreSQL READ COMMITTED вы увидите последнее зафиксированное состояние строк.

Для задачи #42 это всё ещё pending, поскольку Worker A не выполнил COMMIT.

Однако Worker B не получил #42: несмотря на видимый статус pending, эта строка была заблокирована.

Теперь вернитесь в терминал A и нажмите Enter.

Worker A завершит транзакцию, обработает задачу #42 и изменит её статус на completed.

Проверьте итоговое состояние таблицы:

Вывод: два независимых Python-воркера могут резервировать разные задачи, используя общий PostgreSQL.

При этом SKIP LOCKED позволяет второму воркеру не ожидать освобождения уже заблокированной строки.

Эксперимент №3. Ошибка после COMMIT

Цель: исследовать, что произойдёт, если воркер успел зарезервировать задачу, но не смог её выполнить.

Завершите предыдущие учебные транзакции и снова выполните 02_reset.sql.

В файле worker.py временно замените тело функции process_task():

def process_task(task: dict) -> None:
    raise RuntimeError("Ошибка внешнего сервиса")

Запустите:

python worker.py A

Программа завершится с исключением.

Прежде чем проверять таблицу, ответьте:

• Останется ли задача #42 в статусе pending?
• Произойдёт ли откат транзакции резервирования?
• Какую задачу получит следующий воркер?

Теперь проверьте состояние таблицы PostgreSQL.

Задача #42 должна иметь статус processing.

Почему?

Функция claim_task() успешно завершила свою транзакцию до вызова process_task().

Поэтому исключение в Python не отменяет уже зафиксированное изменение статуса задачи.

Верните первоначальную реализацию process_task() и запустите:

python worker.py B

Worker B получит задачу #43, поскольку #42 больше не соответствует условию status = 'pending'.

Вывод: безопасное резервирование не гарантирует успешное выполнение задачи.

Если воркер завершится после COMMIT, задача может остаться в статусе processing.

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

8. Типичные инженерные ошибки

Ошибка №1. Длительная работа внутри транзакции

Рассмотрим код:

with SessionFactory.begin() as session:
    task = session.scalars(stmt).first()

    if task is not None:
        task.status = "processing"

        process_task(task)

Здесь обработка задачи выполняется до завершения транзакции.

Если process_task() работает 30 секунд, транзакция остаётся открытой всё это время.

Строковые блокировки, полученные при резервировании, продолжают удерживаться. Также дольше используется соединение PostgreSQL.

Это увеличивает расход ресурсов и усложняет конкурентную работу.

Для нашего сценария правильнее разделить операции:

task = claim_task()

if task:
    process_task(task)
    mark_as_completed(task["id"])

Ошибка №2. Одна Session на всех воркеров

Не следует создавать общую сессию:

session = SessionFactory()

И затем одновременно использовать этот экземпляр из нескольких обработчиков.

Session содержит состояние текущей транзакции и не предназначена для одновременного использования несколькими потоками.

В нашей реализации каждая функция самостоятельно создаёт отдельную сессию:

with SessionFactory.begin() as session:
    ...

Ошибка №3. Считать COMMIT гарантией выполнения задачи

COMMIT гарантирует фиксацию изменений в PostgreSQL при успешном завершении транзакции.

Но он не означает, что Python успешно выполнил последующую работу.

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

Поэтому надёжная обработка задач требует дополнительных механизмов восстановления и контроля повторных операций.

9. Проверка понимания

Перед итоговым тестированием попробуйте самостоятельно ответить на следующие вопросы.

1. Почему session.flush() не равнозначен COMMIT?
2. Почему второй воркер получает задачу #43, пока первый удерживает блокировку #42?
3. Почему длительную обработку задачи мы выполняем после завершения транзакции?
4. Почему исключение внутри process_task() не отменяет ранее выполненное резервирование?

Если ответы вызывают затруднения, вернитесь к соответствующему эксперименту и попробуйте объяснить поведение PostgreSQL через границы транзакций.

10. Итоги урока

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

Мы научились:

• Создавать Engine и отдельные Session.
• Работать с существующей таблицей через ORM-модель.
• Использовать with_for_update(skip_locked=True).
• Резервировать задачу внутри одной транзакции.
• Различать flush() и COMMIT.
• Разделять резервирование и длительную обработку.
• Проверять конкурентное поведение двух Python-воркеров.
• Объяснять последствия сбоя после фиксации транзакции.

Главный инженерный вывод:

Резервирование задачи должно быть атомарным, а транзакция резервирования — короткой.

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

Теперь у нас есть Python-приложение, в котором несколько независимых воркеров могут конкурентно получать задачи из PostgreSQL.

Следующий шаг — научиться анализировать ограничения этого решения и проектировать механизмы восстановления после отказов.

Дополнительные материалы

• SQLAlchemy 2.x — Session Basics
• SQLAlchemy 2.x — SELECT и with_for_update()
• PostgreSQL 16 — SELECT
• PostgreSQL 16 — Explicit Locking