Hintergrundtasks¶
Sie können Hintergrundtasks (Hintergrund-Aufgaben) definieren, die nach der Rückgabe einer Response ausgeführt werden sollen.
Das ist nützlich für Vorgänge, die nach einem Request ausgeführt werden müssen, bei denen der Client jedoch nicht unbedingt auf den Abschluss des Vorgangs warten muss, bevor er die Response erhält.
Hierzu zählen beispielsweise:
- E-Mail-Benachrichtigungen, die nach dem Ausführen einer Aktion gesendet werden:
- Da die Verbindung zu einem E-Mail-Server und das Senden einer E-Mail in der Regel „langsam“ ist (einige Sekunden), können Sie die Response sofort zurücksenden und die E-Mail-Benachrichtigung im Hintergrund senden.
- Daten verarbeiten:
- Angenommen, Sie erhalten eine Datei, die einen langsamen Prozess durchlaufen muss. Sie können als Response „Accepted“ (HTTP 202) zurückgeben und die Datei im Hintergrund verarbeiten.
BackgroundTasks verwenden¶
Importieren Sie zunächst BackgroundTasks und definieren Sie einen Parameter in Ihrer Pfadoperation-Funktion mit der Typdeklaration BackgroundTasks:
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
FastAPI erstellt für Sie das Objekt vom Typ BackgroundTasks und übergibt es als diesen Parameter.
Eine Taskfunktion erstellen¶
Erstellen Sie eine Funktion, die als Hintergrundtask ausgeführt werden soll.
Es handelt sich schlicht um eine Standard-Funktion, die Parameter empfangen kann.
Es kann sich um eine async def- oder normale def-Funktion handeln. FastAPI weiß, wie damit zu verfahren ist.
In diesem Fall schreibt die Taskfunktion in eine Datei (den Versand einer E-Mail simulierend).
Und da der Schreibvorgang nicht async und await verwendet, definieren wir die Funktion mit normalem def:
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
Den Hintergrundtask hinzufügen¶
Übergeben Sie innerhalb Ihrer Pfadoperation-Funktion Ihre Taskfunktion mit der Methode .add_task() an das Hintergrundtasks-Objekt:
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
.add_task() erhält als Argumente:
- Eine Taskfunktion, die im Hintergrund ausgeführt wird (
write_notification). - Eine beliebige Folge von Argumenten, die der Reihe nach an die Taskfunktion übergeben werden sollen (
email). - Alle Schlüsselwort-Argumente, die an die Taskfunktion übergeben werden sollen (
message="some notification").
Dependency Injection¶
Die Verwendung von BackgroundTasks funktioniert auch mit dem Dependency Injection System. Sie können einen Parameter vom Typ BackgroundTasks auf mehreren Ebenen deklarieren: in einer Pfadoperation-Funktion, in einer Abhängigkeit (Dependable), in einer Unterabhängigkeit usw.
FastAPI weiß, was jeweils zu tun ist und wie dasselbe Objekt wiederverwendet werden kann, sodass alle Hintergrundtasks zusammengeführt und anschließend im Hintergrund ausgeführt werden:
from typing import Annotated
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: str | None = None):
if q:
message = f"found query: {q}\n"
background_tasks.add_task(write_log, message)
return q
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
from typing import Annotated, Union
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
if q:
message = f"found query: {q}\n"
background_tasks.add_task(write_log, message)
return q
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
from typing import Union
from fastapi import BackgroundTasks, Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
if q:
message = f"found query: {q}\n"
background_tasks.add_task(write_log, message)
return q
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
Tipp
Bevorzugen Sie die Annotated-Version, falls möglich.
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: str | None = None):
if q:
message = f"found query: {q}\n"
background_tasks.add_task(write_log, message)
return q
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
Tipp
Bevorzugen Sie die Annotated-Version, falls möglich.
from typing import Union
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
if q:
message = f"found query: {q}\n"
background_tasks.add_task(write_log, message)
return q
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
In obigem Beispiel werden die Nachrichten, nachdem die Response gesendet wurde, in die Datei log.txt geschrieben.
Wenn im Request ein Query-Parameter enthalten war, wird dieser in einem Hintergrundtask in das Log geschrieben.
Und dann schreibt ein weiterer Hintergrundtask, der in der Pfadoperation-Funktion erstellt wird, eine Nachricht unter Verwendung des Pfad-Parameters email.
Technische Details¶
Die Klasse BackgroundTasks stammt direkt von starlette.background.
Sie wird direkt in FastAPI importiert/inkludiert, sodass Sie sie von fastapi importieren können und vermeiden, versehentlich das alternative BackgroundTask (ohne das s am Ende) von starlette.background zu importieren.
Indem Sie nur BackgroundTasks (und nicht BackgroundTask) verwenden, ist es dann möglich, es als Pfadoperation-Funktion-Parameter zu verwenden und FastAPI den Rest für Sie erledigen zu lassen, genau wie bei der direkten Verwendung des Request-Objekts.
Es ist immer noch möglich, BackgroundTask allein in FastAPI zu verwenden, aber Sie müssen das Objekt in Ihrem Code erstellen und eine Starlette-Response zurückgeben, die es enthält.
Weitere Details finden Sie in der offiziellen Starlette-Dokumentation für Hintergrundtasks.
Vorbehalt¶
Wenn Sie umfangreiche Hintergrundberechnungen durchführen müssen und diese nicht unbedingt vom selben Prozess ausgeführt werden müssen (z. B. müssen Sie Speicher, Variablen, usw. nicht gemeinsam nutzen), könnte die Verwendung anderer größerer Tools wie z. B. Celery von Vorteil sein.
Sie erfordern in der Regel komplexere Konfigurationen und einen Nachrichten-/Job-Queue-Manager wie RabbitMQ oder Redis, ermöglichen Ihnen jedoch die Ausführung von Hintergrundtasks in mehreren Prozessen und insbesondere auf mehreren Servern.
Um ein Beispiel zu sehen, sehen Sie sich die Projektgeneratoren an. Sie alle enthalten Celery, bereits konfiguriert.
Wenn Sie jedoch über dieselbe FastAPI-Anwendung auf Variablen und Objekte zugreifen oder kleine Hintergrundtasks ausführen müssen (z. B. das Senden einer E-Mail-Benachrichtigung), können Sie einfach BackgroundTasks verwenden.
Zusammenfassung¶
Importieren und verwenden Sie BackgroundTasks mit Parametern in Pfadoperation-Funktionen und Abhängigkeiten, um Hintergrundtasks hinzuzufügen.