FastAPI¶
Ультрамодный фреимворк на котором все щас пишут круды и чего посложнее
Рецепты¶
Синхронный или асинхронный¶
Наверняка знаете, что роут можно сделать либо асинхронным, либо синхронным. Так когда какой надо делать?
Скорее всего первая мысль которая придет вам в голову будет звучать как-то так - если у нас есть I/O-bound задачи (например работа с БД), то надо использовать асинхронщину, если всё остальное - потоки, процессы и так далее. Но тут есть несколько нюансов:
- Под капотом FastAPI отлично справляется с обработкой как синхронных, так и асинхронных роутов. Если роут асинхронный, то задача по его обработке запустится в event loop, если синхронный - то в thread pool.
- Так как синхронные роуты запускаются в thread pool, иногда просто нет вообще никакого смысла тащить в проект асинхронную ORM, так как всё и так будет работать не блокируя основное приложение.
Возьмем вот такой роут:
@router.get("/nonblocking-sync-operation")
def nonblocking_sync_operation():
time.sleep(10)
return {"test": "test"}
После того как мы перейдем по этому роуту, мы будем ждать 10 секунд и в конце получим ответ. При этом сам FastAPI не заблокируется, и сможет обрабатывать другие подключения - потому что функция запустилась в отдельном потоке.
А теперь возьмем вот такой роут:
@router.get("/blocking-sync-operation")
async def blocking_sync_operation():
time.sleep(10)
return {"test": "test"}
Здесь после перехода по роуту функция запустится в event loop и sleep заблокирует всё приложение до тех пор, пока он не пройдет. То есть, FastAPI вообще перестанет принимать подключения до тех пор, пока функция не выполнится.
Как запускать синхронные функции в асинхронном роуте FastAPI?¶
Иногда так случается, что приходится использовать синхронный код в асинхронном роуте. Если мы попытаемся вызвать синхронную функцию в асинхронном коде - наш event loop заблокируется и всё "зависнет" до тех пор, пока синхронный код не отработает.
Решений, как это сделать, на самом деле много. Самый простой вариант, который предоставляет FastAPI (а если быть точнее - Starlette, который использует anyio) - функция run_in_threadpool, которая запустит синхронный код в потоке:
@app.get("/")
async def my_router():
result = await service.execute()
client = SyncClient()
return await run_in_threadpool(client.execute, data=result)
Кстати, BackgroundTask использует тот же самый способ, только он не возвращает результат выполнения.
Изменение типов в рантайме для документации¶
Если вам понадобится в рантайме поменить тип принимаемого объекта в роуте FastAPI, вот вам небольшой рецепт.
Для одного проекта я решил сделать плагин-систему, которая автоматически регистрирует python-файлы и добавляет эндпоинты в API написанном на FastAPI. Так как я использую OpenAPI, очень хочется, чтобы принимаемые и возвращаемые типы отображались в нём корректно.
Допустим, есть у меня вот такой код, создающий эндпоинты в зависимости от переданного Action:
class Action[ReqT, RespT](Protocol):
name: str
request_schema: Type[ReqT] # тип который мы будем принимать
response_schema: Type[RespT] # тип который мы будем возвращать
interact: Callable[[ReqT], Coroutine[Any, Any, RespT]]
def create_action_handler(action: Action):
# какой-то код
async def action_endpoint(entity_id: int, data: dict[str, Any]) -> Any:
# какой-то код
return await action.interact(server, data)
return action_endpoint
def add_action_route(router: APIRouter, action: Action):
handler = create_action_handler(action)
router.post(
f"/plugins/{action.name}",
response_model=action.response_schema, # установим схему для ответа
)(handler)
Схему для ответа я установил при регистрации нашего эндпоинта в response_model. А что делать с data, тип которого должен быть не dict[str, Any] а action.request_schema?
Для начала, я подумал, что раз FastAPI строит доку из аннотаций, наверное я смогу поменять аннотации и на этом дело сделано:
Но это не помогло - аннотации поменялись, а вот схема - нет!Оказывается, FastAPI не использует аннотации, а использует запеченные Signature, которые собираются из Dependency. При чем, в понимании FastAPI, dependency не только источник каких-то функций, а еще и наших схем. А дальше он из них и строит OpenAPI.
Зная это нам остается только одно - изменить signature перед регистрацией роута:
sig = inspect.signature(handler) # достаем сигнатуру
new = []
for p in sig.parameters.values():
if p.name == "data": # находим нашу data
p = p.replace(annotation=action.request_schema) # меняем
new.append(p)
handler.__signature__ = sig.replace(parameters=new) # записываем новую сигнатуру
Вуаля! Теперь в OpenAPI будет указанная в action.request_schema схема.