Используйте SDK в приложении на FastAPI или Django
PythonОдин клиент на рабочий процесс, его копия, которая пишет журнал с идентификатором каждого запроса, запуски, переданные Slen без ожидания в обработчике, и клиент, который никогда не переходит через fork.
6 мин чтения
Эта статья — о SDK для Python.
У Python SDK нет своей интеграции с фреймворком: приложение на FastAPI или Django собирает клиент и вызывает его, как любой другой код. Серверу нужно место, где собирается клиент, его копия для каждого запроса и обработчик, который не ждёт запуска. Эта страница показывает оба фреймворка; тесты SDK запускают их код в тестовом клиенте каждого.
Один клиент на процесс
Сервер работает в нескольких рабочих процессах, и каждый обслуживает много запросов. Соберите в каждом процессе один клиент и обслуживайте им каждый запрос этого процесса.
Клиент хранит свои соединения и настройки, но никогда — состояние вызова, поэтому один клиент обслуживает каждый запрос своего процесса — его нити и задачи одновременно.
Клиент собирают в том процессе, который им пользуется, и никогда — при импорте модуля: клиент, собранный до того, как сервер породит рабочие процессы через fork, делил бы между ними свои соединения, — ниже.
AsyncSlenпринадлежит циклу событий, в котором его собрали, поэтому сервер на asyncio собирает его внутри этого цикла.
FastAPI
import logging
from collections.abc import AsyncGenerator, MutableMapping
from contextlib import asynccontextmanager
from typing import Any
from fastapi import FastAPI, Request
from slen import AsyncSlen
# One client per worker process, built in the worker's own event loop when it starts and closed when it stops —
# never when the module is imported, so workers forked from one process share no connections.
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
async with AsyncSlen() as slen:
app.state.slen = slen
yield
app = FastAPI(lifespan=lifespan)
# Keeps the fields of each event of the SDK and adds the id of the request being served.
class RequestLog(logging.LoggerAdapter[logging.Logger]):
def process(self, msg: Any, kwargs: MutableMapping[str, Any]) -> tuple[Any, MutableMapping[str, Any]]:
return msg, {**kwargs, "extra": {**(kwargs.get("extra") or {}), **(self.extra or {})}}
def slen_of(request: Request) -> AsyncSlen:
# A copy of the worker's client that logs with this request's id; the shared client does not change.
shared: AsyncSlen = request.app.state.slen
log = RequestLog(logging.getLogger("orders"), {"http_request_id": request.headers.get("x-request-id", "")})
return shared.with_logger(log)Клиент собирается в
lifespanприложения, который каждый рабочий процесс выполняет в своём цикле событий при запуске, аasync withзакрывает его, когда процесс останавливается.slen_of()даёт каждому запросу копию клиента — Журнал каждого запроса.
@app.post("/orders/{order_id}/email", status_code=202)
async def email_the_order(order_id: int, request: Request) -> dict[str, str]:
# The run goes to a worker and the request is answered at once: a handler does not wait for a run.
run = await slen_of(request).operations.execute_async(
"generate_email",
{"messages": [{"role": "user", "content": f"Write a short email: order {order_id} has shipped."}]},
idempotency_key=f"order-{order_id}:email",
)
return {"execution_id": run.execution_id, "status": run.status}@app.get("/runs/{execution_id}")
async def run_status(execution_id: str, request: Request) -> dict[str, str]:
# One read of the run; its end arrives as the execution.finished webhook.
run = await slen_of(request).executions.get(execution_id)
return {"execution_id": run.execution_id, "status": run.status}Django
import functools
import logging
from collections.abc import MutableMapping
from typing import Any
from django.http import HttpRequest, JsonResponse
from slen import Slen
# One client per worker process, built by the first request the process serves — never when the module is
# imported or the app is ready, so workers forked after loading the application share no connections.
@functools.cache
def shared_slen() -> Slen:
return Slen()
# Keeps the fields of each event of the SDK and adds the id of the request being served.
class RequestLog(logging.LoggerAdapter[logging.Logger]):
def process(self, msg: Any, kwargs: MutableMapping[str, Any]) -> tuple[Any, MutableMapping[str, Any]]:
return msg, {**kwargs, "extra": {**(kwargs.get("extra") or {}), **(self.extra or {})}}
def slen_of(request: HttpRequest) -> Slen:
# A copy of the process's client that logs with this request's id; the shared client does not change.
log = RequestLog(logging.getLogger("orders"), {"http_request_id": request.headers.get("X-Request-Id", "")})
return shared_slen().with_logger(log)shared_slen()собирает клиент на первом запросе, который обслуживает процесс, аfunctools.cacheхранит его для остальных — не при импорте модуля и не тогда, когда приложение готово: сервер, который загружает приложение до того, как породит рабочие процессы, делает это до fork.Представление Django синхронное, поэтому оно вызывает
Slen.
def email_the_order(request: HttpRequest, order_id: int) -> JsonResponse:
# The run goes to a worker and the request is answered at once: a view does not wait for a run.
run = slen_of(request).operations.execute_async(
"generate_email",
{"messages": [{"role": "user", "content": f"Write a short email: order {order_id} has shipped."}]},
idempotency_key=f"order-{order_id}:email",
)
return JsonResponse({"execution_id": run.execution_id, "status": run.status}, status=202)def run_status(request: HttpRequest, execution_id: str) -> JsonResponse:
# One read of the run; its end arrives as the execution.finished webhook.
run = slen_of(request).executions.get(execution_id)
return JsonResponse({"execution_id": run.execution_id, "status": run.status})from django.urls import path
from django.views.decorators.http import require_GET, require_POST
urlpatterns = [
path("orders/<int:order_id>/email", require_POST(email_the_order)),
path("runs/<str:execution_id>", require_GET(run_status)),
]Журнал каждого запроса
Журнал SDK идёт в один логгер на весь процесс. Чтобы найти записи одного запроса, with_logger() даёт каждому запросу копию общего клиента, которая пишет в свой логгер. Копия отправляет запросы через те же соединения и записывает спаны тем же записывающим; своих соединений у неё нет, а общий клиент остаётся каким был — Журнал.
RequestLogдобавляет к каждой записи идентификатор запроса, который обслуживает ваш сервер, и сохраняет поля, которые несёт каждое событие SDK. Обычныйlogging.LoggerAdapterдо Python 3.13 заменяетextraвызова своим, и поля события потерялись бы. С 3.13 обычный адаптер сmerge_extra=Trueобъединяет их сам; примеры оставляютRequestLog, потому что SDK работает и на 3.11 и 3.12.Ваш идентификатор идёт под
http_request_id.request_idзаписи SDK — идентификатор ответа Slen, тот, по которому его поддержка находит запрос.
Не ждите в обработчике
executions.wait() держит запрос вашего вызывающего открытым, пока длится запуск, и всё это время занимает нить или задачу рабочего процесса. Обработчик передаёт запуск Slen через operations.execute_async() и сразу отвечает 202. Окончание запуска приходит вебхуком execution.finished — Вебхуки, — или его ждёт задача вашей очереди. Чтение статуса, executions.get(), — один запрос, и ему место в обработчике.
Запуск, которого обработчик всё же ждёт, operations.execute(), получает свой бюджет: timeout и retries вызова, а не минуты клиента — Повторы и таймауты.
Закройте клиент
async with AsyncSlen() as slen: в lifespan закрывает соединения клиента, когда рабочий процесс останавливается, а close() делает то же там, где такого блока нет. Копия, сделанная with_logger(), своих соединений не имеет, поэтому её закрытие ничего не закрывает. httpx.Client, который вы передали как http_client, остаётся открытым: закрываете его вы.
Fork и --preload
Клиент, собранный до fork, делит свои соединения с каждым потомком, и тогда процессы читают и пишут одни и те же сокеты. gunicorn --preload и любой сервер, который загружает приложение до того, как породит рабочие процессы, делают это с клиентом, собранным при импорте модуля. Собирайте клиент в рабочем процессе, как оба примера. В другой процесс клиент тоже не переходит: его сериализация через pickle или копирование выбрасывает TypeError.
Что читать дальше
Конфигурация — опции клиента и
AsyncSlen.Асинхронные запуски и ожидание — окончание запуска без ожидания.
Журнал — события журнала и их поля.
Вебхуки — примите
execution.finished.