5) FAPI+FonPr/main.py - main - K_Msk/FastAPI_First

5) FAPI+FonPr/main.py
61 строка3 KB
Skrppapa
Fon
16 сен 2025, 21:57
16 сен 2025, 21:57ec4aaa2
from fastapi import FastAPI, BackgroundTasks
import time
import asyncio

app = FastAPI()


# Имитация действия и ответ
def sync_task():
    time.sleep(3)
    print("Отправлен email")

# Асинхронная имитация действия и ответ
async def async_task():
    await asyncio.sleep(3)
    print("Сделан запрос в сторонний API")

# Если используем функцию в ручке - фоновой она не будет. Мы будем ждать 3 секнды, даже если функция ассинхронная
# Задача - сделать эту функцию реально фоновой. Юзер получает уведомление сразу "Будет сделано, можешь дальше пользоваться сервисом" то есть не ждать эти 3 условные секнуды

# Как работает FAPI - Если все ручки асинхронные, то это один event_loop (событийный цикл) он будет управлять какая ручка сейчас выполняется, следить за ними.
# Если есть синхронная ручка - то при использовании создается новый отдельный поток (это ограничивает количесвто одновременных запросов к API)
# Значит, что нам нужно? Положить задачу в событийный цикл и продолжить работу с сервисом
# Ручка ниже именно ждет выполнения.
#
#@app.post("/")
#async def some_route():
#    ...
#    await async_task()
#    return({'ok': True})
#
# Что делаем: Что бы вызвать асинхронную функцию, нужно писать await
# Но ниже я просто создаю таску и не пишу await
# В результате ответ пользователю приходит сразу. А сама задача выполняется пофакту чуть позже.
# Это работает с асинхронными функциями

@app.post("/")
async def some_route():
    ...
    asyncio.create_task(async_task())
    return({'ok': True})


# Для синхронных ручек импортируем BackgroundTasks
# Действуем схожим образом с асинхронной ручкой
# Тут уже мы передаем аргумент и говорим что тип данных у него BackgroundTasks - это важно
@app.post("/sync")
def some_route_sync(bg_tasks: BackgroundTasks):
    ...
    bg_tasks.add_task(sync_task) # <- Обрати внимание функция не нужно вызывать (пишем без ())
    return({'ok': True})




if __name__ == '__main__':
    uvicorn.run('main:app')

# Активация . fonenv\Scripts\activate

# uvicorn main:app --reload