Comment exécuter du code synchrone bloquant (par exemple, un calcul d'une seconde) via un endpoint dans FastAPI, sans bloquer la boucle d'événements ?

В FastAPI, чтобы выполнить синхронный блокирующий код (например, вычисления, которые занимают 1 секунду) без блокировки основного EventLoop, нужно запускать этот код в отдельном потоке или процессе. Для этого используется функция run_in_executor из модуля asyncio.

Пример:

import time
import asyncio
from fastapi import FastAPI

app = FastAPI()

def blocking_task():
    time.sleep(1)  # блокирующая операция
    return "Готово"

@app.get("/sync-task")
async def run_blocking():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, blocking_task)
    return {"result": result}

Здесь run_in_executor запускает blocking_task в пуле потоков по умолчанию, не блокируя EventLoop, и позволяет обрабатывать другие запросы параллельно.