← назад к разделу

Предыдущие статьи раздела разбирали asyncio по частям. Здесь они складываются в один обработчик FastAPI, каким он выглядит в сервисе: с клиентами из lifespan, зависимостями, фоновыми задачами, корректной остановкой и метриками. Факты про порядок выполнения проверены на FastAPI 0.142 и Starlette 1.7, потому что именно порядок чаще всего расходится с ожиданиями.

Обязательно

Каркас: lifespan с клиентами и фоновыми задачами

Всё долгоживущее создаётся один раз в lifespan и живёт в app.state: клиенты, пулы, группа фоновых задач. Порядок закрытия обратный порядку создания, и сначала останавливаются потребители, потом закрываются клиенты, которыми они пользовались.

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(base_url=settings.catalog_url, timeout=httpx.Timeout(2.0, connect=0.5))
    app.state.pg = await asyncpg.create_pool(settings.pg_dsn, min_size=5, max_size=20, command_timeout=10)
    async with anyio.create_task_group() as tg:
        tg.start_soon(consume_orders, app.state.pg)
        try:
            yield
        finally:
            tg.cancel_scope.cancel()
    await app.state.pg.close()
    await app.state.http.aclose()

Фоновый потребитель внутри группы задач: его падение поднимется в lifespan и остановит приложение честно, а не оставит сервис без потребителя. При остановке группа отменяет потребителя и ждёт его, и только после этого закрываются пул и клиент. Что должен сделать потребитель по отмене, разобрано в статье про остановку потребителя Kafka.

Обработчик: async def, зависимости и сессия

async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
    async with request.app.state.session_factory() as session, session.begin():
        yield session

@router.post("/orders/{order_id}/pay", status_code=202)
async def pay_order(order_id: int, body: PayRequest, session: AsyncSession = Depends(get_session),
                    http: httpx.AsyncClient = Depends(get_http)) -> PayAccepted:
    async with asyncio.timeout(1.5):
        order = await OrderRepo(session).get_for_update(order_id)
        order.pay(body.amount)
        await Outbox(session).add(OrderPaid.from_order(order))
    return PayAccepted(order_id=order_id)

Четыре решения в этих строках. Обработчик async def, потому что всё внутри ждёт сеть асинхронно; def отправил бы его в пул на 40 потоков. Транзакция открыта зависимостью и закрывается в её коде после yield: commit на выходе, rollback при исключении. Внешних вызовов внутри транзакции нет: событие уходит через outbox, а не напрямую в Kafka. И весь обработчик под asyncio.timeout, чтобы бюджет ответа держался даже при медленной базе.

Когда выполняется код после yield

Это место, где интуиция подводит. Проверено на FastAPI 0.142: по умолчанию код зависимости после yield выполняется после того, как ответ отправлен клиенту, и после фоновых задач. Для сессии базы это значит: клиент получил 202, а commit ещё не произошёл. Если следующий запрос клиента сразу читает заказ, он может не увидеть оплату. Параметр Depends(get_session, scope="function") меняет порядок: код после yield выполняется до отправки ответа, и клиент получает 202 уже после commit; проверено, что при scope="function" ответ задерживается на время кода после yield.

Правило: для зависимостей, которые фиксируют транзакцию или освобождают ресурс, от которого зависит видимость результата, scope="function"; для зависимостей, которые только закрывают что-то своё (временный файл, трассировочный спан), подходит умолчание. И второе следствие: исключение в коде после yield при умолчании уже не изменит ответ, он отправлен; ошибка commit попадёт в лог, а клиент увидит успех. Ещё одна причина фиксировать до ответа.

BackgroundTasks и настоящие фоновые задачи

BackgroundTasks выполняются после отправки ответа, в той же задаче запроса: async def прямо в цикле, def в пуле потоков. Это подходит для короткой работы, результат которой клиенту не нужен: отправить письмо, записать метрику. Три ограничения. Фоновая задача выполняется в пределах жизни задачи запроса, и если процесс останавливается, uvicorn ждёт её как часть запроса, но не дольше таймаута завершения. Ошибка в ней попадает в лог, клиент о ней не узнает. И ей нельзя отдавать сессию запроса: с умолчанием scope сессия ещё открыта во время фоновой задачи, но закроется сразу после неё, а с scope="function" уже закрыта; фоновая задача открывает свою сессию и получает идентификаторы.

Работа, которая должна пережить запрос и перезапуск (повторы, долгие операции), не место для BackgroundTasks: её кладут в outbox или очередь, а выполняет отдельный потребитель. Подробнее в статье про фоновые задачи FastAPI.

Клиент ушёл

Проверено на Starlette 1.7: если клиент закрыл соединение, обработчик не отменяется и работает до конца. Для короткого обработчика это правильно: транзакция завершится, ответ просто некому отдать. Для долгого (отчёт на минуту, потоковая выдача) это трата ресурсов, и обработчик сам проверяет await request.is_disconnected() между этапами и прекращает работу. В потоковых ответах разрыв соединения проявляется исключением при записи в поток, и генератор должен освобождать ресурсы в finally.

Остановка

Порядок при SIGTERM: uvicorn перестаёт принимать соединения, ждёт обработчики в полёте до --timeout-graceful-shutdown, выполняет код после yield в lifespan (отмена фоновых задач, закрытие клиентов), затем asyncio.run отменяет всё, что осталось. Три настройки, которые делают это предсказуемым: --timeout-graceful-shutdown меньше terminationGracePeriodSeconds пода, чтобы под не убили по SIGKILL на середине; таймауты на всех внешних ожиданиях, иначе обработчик в полёте не завершится к сроку; и отсутствие проглоченных CancelledError в фоновых задачах. Как это совместить с preStop и проверками готовности, разбирает статья про завершение в Kubernetes.

Флаги uvicorn, которые относятся к конкурентности

--workers N даёт N процессов с отдельными циклами; лимиты и пулы умножаются на N. --limit-concurrency M отвечает 503 при более чем M одновременных соединениях на процесс, грубая защита от перегрузки без собственной прослойки. --timeout-keep-alive ограничивает простой соединений, --backlog очередь входящих. --loop uvloop подключает более быструю реализацию цикла, если пакет установлен; выигрыш заметен на мелких запросах и несуществен, когда время уходит в базу.

Что мерить

Пять метрик, по которым видно состояние цикла событий и его границ. Задержка планирования: как долго готовая задача ждёт очереди; измеряется задачей, которая каждые 100 миллисекунд спит и сравнивает ожидаемое и фактическое время пробуждения, и её рост означает блокировку цикла. Число живых задач из asyncio.all_tasks(): рост без нагрузки это утечка. Занятость пулов: соединения в использовании и ожидание выдачи для asyncpg и httpx. Доля запросов, завершённых по таймауту, по каждому соседу. И длительность запросов в пуле потоков, если to_thread в коде есть. Как отдавать их в Prometheus, рассказывает статья про метрики на Python.

async def loop_lag_probe(histogram, interval: float = 0.1) -> None:
    loop = asyncio.get_running_loop()
    while True:
        expected = loop.time() + interval
        await asyncio.sleep(interval)
        histogram.observe(max(0.0, loop.time() - expected))
Дополнительно: при первом чтении можно пропустить

Глубже: порядок middleware и где ставить таймаут на весь запросрасширенное

Middleware выполняются в порядке, обратном регистрации: последнее добавленное первым получает запрос. Проверено на Starlette 1.7, и это важно для двух вещей. Таймаут на весь запрос ставят во внешнем middleware, чтобы он покрывал и остальные; а код после await call_next(request) не выполняется при исключении в обработчике, поэтому метрики и очистку пишут в finally. Middleware на базе BaseHTTPMiddleware выполняет обработчик в отдельной задаче и ограничивает потоковые ответы, поэтому для таймаута на запрос и для отмены по уходу клиента надёжнее чистое ASGI-middleware, которое работает с scope, receive и send напрямую и видит событие http.disconnect. Как устроен этот слой, разбирает статья про middleware и ошибки во FastAPI.

И про зависимости с yield и TestClient: в тестах через ASGITransport порядок «ответ, потом код после yield» сохраняется, поэтому тест, который сразу после запроса читает базу через другую сессию, может не увидеть commit при умолчании scope. Это та же ошибка, что и у реального клиента, и она же лечится scope="function" для сессии.

Коротко

  • Долгоживущее в lifespan и app.state: клиенты, пулы, группа фоновых задач; закрытие в обратном порядке после отмены потребителей.
  • Обработчик async def с сессией из yield-зависимости, транзакцией без внешних вызовов, outbox для событий и asyncio.timeout на бюджет ответа.
  • Код после yield по умолчанию выполняется после отправки ответа и после фоновых задач (FastAPI 0.142); для commit нужен Depends(..., scope="function"), иначе клиент получает ответ до фиксации, а ошибка фиксации не меняет ответ.
  • BackgroundTasks выполняются после ответа в задаче запроса, ошибки только в логе, своя сессия; работа, которая должна пережить запрос, идёт в outbox или очередь.
  • Уход клиента обработчик не отменяет (Starlette 1.7): долгие обработчики проверяют request.is_disconnected().
  • Остановка: --timeout-graceful-shutdown меньше срока пода, таймауты на ожиданиях, отмена без проглатывания; --limit-concurrency и --workers как грубые рычаги.
  • Метрики: задержка планирования цикла, число задач, занятость пулов, доля таймаутов по соседям, длительность в пуле потоков.
  • Middleware выполняются в обратном порядке регистрации, код после call_next при исключении пропускается; таймаут на запрос и уход клиента ловят в ASGI-middleware.

Что почитать дальше