Django бест-практикс у нас были, а вот по FastAPI не было. А теперь есть :)

#fastapi

FastAPI обновился до 0.89.0 и сделал фичу которой давно не хватало - наконец возвращаемый тип можно писать в аннотациях. Раньше, чтобы OpenAPI дока по нему сгенерировалась, надо было делать так:

@app.get("/", response_model=Model)
def endpoint():
    return Model(name="Yurii")

Теперь же можем делать вот так:

@app.get("/")
def endpoint() -> Model:
    return Model(name="Yurii")

Мне до сих пор не понятно, почему так не сделали раньше. Возможно автору такой метод виделся удобным.
Кому интересны детали - вот PR.

#fastapi

Немного про роутинг в FastAPI

Если вы пользовались FastAPI, то наверняка знаете, что роут можно сделать либо асинхронным, либо синхронным. Так когда какой надо делать?

Скорее всего первая мысль которая придет вам в голову будет звучать как-то так - если у нас есть I/O-bound задачи (например работа с БД), то надо использовать асинхронщину, если всё остальное - потоки, процессы и так далее. Но тут есть несколько нюансов:

1) Под капотом FastAPI отлично справляется с обработкой как синхронных, так и асинхронных роутов. Если роут асинхронный, то задача по его обработке запустится в event loop, если синхронный - то в thread pool.
2) Так как синхронные роуты запускаются в 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 небольшой CRUD и вы думаете тащить асинхронную ORM - задумайтесь, а надо ли она вам там вообще?

Ссылки:
- Path operation functions

#fastapi

А я к вам с новостями.

FastAPI в версии 0.100.0-beta1 поддерживает Pydantic v2 в бета-режиме. Да-да, это тот самый Pydantic, внутренности которого написаны на Rust. Гайд по миграции можно почитать здесь, а релиз тут.

#fastapi #pydantic

Как запускать синхронные функции в асинхронном роуте 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 #anyio