FastAPI ist ein modernes, schnelles (hochleistungsfähiges) Internet-Framework zum Erstellen von APIs mit Python 3.6+ basierend auf Customary-Python-Typ-Hinweisen. Aufgrund seiner Effizienz, Einfachheit und einer Reihe einzigartiger Funktionen, die es Entwicklern ermöglichen, in Rekordzeit robuste APIs zu erstellen, gewinnt es in der Softwareentwicklungsbranche schnell an Bedeutung. In diesem Artikel konzentrieren wir uns auf eine der leistungsstärksten Funktionen von FastAPI: Hintergrundaufgaben. Insbesondere veranschaulichen wir, wie Sie diese Funktion nutzen können, um Berechnungen asynchron durchzuführen, während die API weiterhin auf andere eingehende Anforderungen reagiert.
Betrachten wir für dieses Beispiel eine FastAPI-App, die Inferenzanfragen verarbeitet. Unsere API akzeptiert eine Eingabenutzlast, plant die Ausführung einer Verarbeitungsaufgabe im Hintergrund und antwortet sofort mit einem Offset für diese Aufgabe. Über diesen Offset kann das Ergebnis der Verarbeitungsaufgabe später abgerufen werden. Diese Artwork der Einrichtung ist in Szenarien üblich, in denen Rechenaufgaben viel Zeit in Anspruch nehmen und wir den Shopper nicht auf den Abschluss der Aufgabe warten lassen möchten.
Lassen Sie uns die Schlüsselkomponenten des Codes aufschlüsseln:
import logging
import time
from typing import Recordimport uvicorn
from fastapi import BackgroundTasks, FastAPI, HTTPException
from pydantic import BaseModel, Area
app = FastAPI()
logging.basicConfig(degree=logging.INFO)
class InferenceRequest(BaseModel):
deployment: str = Area(..., instance="Docker")
framework: str = Area(..., instance="FastAPI")
fn: str = Area(..., instance="default_fn")
class InferenceResponse(BaseModel):
offset: int
stack = []
Zuerst importieren wir die notwendigen Pakete und richten eine grundlegende FastAPI-Anwendung ein. Wir definieren zwei Pydantic-Modelle: InferenceRequest Und InferenceResponse, um das Anforderungs- und Antwortschema für unsere API anzugeben. Wir deklarieren auch eine globale Liste, stackum die Ergebnisse unserer Hintergrundaufgaben zu verfolgen.
def course of(offset) -> listing[str]:
for m in ["result1", "result2", "result3"]:
time.sleep(10)
stack[offset].append(m)stack[offset].append("performed")
Als nächstes definieren wir a course of Funktion, die als Ersatz für unsere zeitaufwändige Rechenaufgabe dient. Es wartet lediglich eine Zeit lang und sendet dann eine Reihe von Ergebnissen an die stack zu einem gegebenen Zeitpunkt offset.
@app.publish("/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: course of(offset))
return InferenceResponse(offset=offset)@app.get("/messages/{offset}", response_model=Record[str])
async def get_messages(offset: int):
if offset < 0 or offset >= len(stack):
increase HTTPException(status_code=404, element="Offset not discovered")
return stack[offset]
Abschließend definieren wir zwei Endpunkte: /schedule Und /messages/{offset}. Der /schedule Endpunkt akzeptiert eine InferenceRequestfügt dem Hintergrund eine neue Aufgabe hinzu background_tasks.add_task()und antwortet sofort mit einem InferenceResponse enthält die offset. Der /messages/{offset} Der Endpunkt ruft die Ergebnisse ab stack für ein gegebenes offset.
In diesem Beitrag haben wir gelernt, wie man FastAPIs verwendet BackgroundTasks um zeitaufwändige Aufgaben so zu planen, dass sie im Hintergrund ausgeführt werden, während die API reaktionsfähig bleibt. Diese Funktion ermöglicht die Erstellung effizienter, leistungsstarker APIs, die Aufgaben unterschiedlicher Dauer bewältigen können, ohne die Hauptanwendung zu blockieren. Viel Spaß beim Codieren!
