Как да използвате BackgroundTasks в FastAPI

FastAPI е модерна, бърза (високопроизводителна) уеб рамка за изграждане на API с Python 3.6+ въз основа на стандартни подсказки от тип Python. Той бързо набира популярност в индустрията за разработка на софтуер поради своята ефективност, простота и набор от уникални функции, които позволяват на разработчиците да създават стабилни API за рекордно кратко време. В тази статия ще се съсредоточим върху една от най-мощните функции на FastAPI: Background Tasks. По-конкретно, ще илюстрираме как да използвате тази функция, за да извършвате изчисления асинхронно, докато API остава отзивчив към други входящи заявки.

Преглед на кода

За този пример нека разгледаме приложение FastAPI, което обработва заявки за извод. Нашият API ще приема входен полезен товар, ще планира изпълнение на задача за обработка във фонов режим и ще реагира незабавно с отместване за тази задача. Резултатът от задачата за обработка може да бъде извлечен по-късно с помощта на това отместване. Този тип настройка е често срещана в сценарии, при които изчислителните задачи отнемат значително време и не искаме да караме клиента да чака задачата да завърши.

Нека разбием ключовите компоненти на кода:

import logging
import time
from typing import List

import uvicorn
from fastapi import BackgroundTasks, FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()

logging.basicConfig(level=logging.INFO)


class InferenceRequest(BaseModel):
    deployment: str = Field(..., example="Docker")
    framework: str = Field(..., example="FastAPI")
    fn: str = Field(..., example="default_fn")


class InferenceResponse(BaseModel):
    offset: int


stack = []

Първо импортираме необходимите пакети и настройваме основно FastAPI приложение. Ние дефинираме два Pydantic модела, InferenceRequest и InferenceResponse, за да уточним схемата за заявка и отговор за нашия API. Ние също така декларираме глобален списък, stack, за да следим резултатите от нашите фонови задачи.

def process(offset) -> list[str]:
    for m in ["result1", "result2", "result3"]:
        time.sleep(10)
        stack[offset].append(m)

    stack[offset].append("done")

След това дефинираме функция process, която е заместник за нашата трудоемка изчислителна задача. Той просто изчаква период и след това изпраща набор от резултати към stack при даден offset.

@app.post("/schedule", response_model=InferenceResponse)
async def inference(request: InferenceRequest, background_tasks: BackgroundTasks):
    start_time = time.time()
    stack.append([])
    offset = len(stack) - 1
    background_tasks.add_task(lambda: process(offset))
    return InferenceResponse(offset=offset)


@app.get("/messages/{offset}", response_model=List[str])
async def get_messages(offset: int):
    if offset < 0 or offset >= len(stack):
        raise HTTPException(status_code=404, detail="Offset not found")
    return stack[offset]

Накрая дефинираме две крайни точки: /schedule и /messages/{offset}. Крайната точка /schedule приема InferenceRequest, добавя нова задача на заден план с помощта на background_tasks.add_task() и незабавно отговаря с InferenceResponse, съдържащо offset. Крайната точка /messages/{offset} извлича резултатите от stack за даден offset.

Заключение

В тази публикация научихме как да използваме BackgroundTasks на FastAPI, за да планираме времеемки задачи, които да се изпълняват във фонов режим, като същевременно поддържаме API отзивчив. Тази функция позволява изграждането на ефективни, високопроизводителни API, които могат да се справят със задачи с различна продължителност, без да блокират основното приложение. Приятно кодиране!