Tareas en Segundo Plano¶

Puedes definir tareas en segundo plano para que se ejecuten después de devolver una response.

Esto es útil para operaciones que necesitan ocurrir después de una request, pero que el cliente no tiene realmente que estar esperando a que la operación termine antes de recibir la response.

Esto incluye, por ejemplo:

• Email notifications sent after performing an action:
  • Como conectarse a un servidor de email y enviar un email tiende a ser "lento" (varios segundos), puedes devolver la response inmediatamente y enviar la notificación por email en segundo plano.
• Processing data:
  • Por ejemplo, digamos que recibes un archivo que debe pasar por un proceso lento, puedes devolver una response de "Accepted" (HTTP 202) y procesar el archivo en segundo plano.

Usando BackgroundTasks¶

Primero, importa BackgroundTasks y define un parámetro en tu función de path operation con una declaración de tipo 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 creará el objeto de tipo BackgroundTasks por ti y lo pasará como ese parámetro.

Crear una función de tarea¶

Crea una función para ejecutar como la tarea en segundo plano.

Es simplemente una función estándar que puede recibir parámetros.

Puede ser una función async def o def normal, FastAPI sabrá cómo manejarla correctamente.

En este caso, la función de tarea escribirá en un archivo (simulando el envío de un email).

Y como la operación de escritura no usa async y await, definimos la función con def normal:

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"}

Añadir la tarea en segundo plano¶

Dentro de tu función de path operation, pasa tu función de tarea al objeto de background tasks con el método .add_task():

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() recibe como argumentos:

• Una función de tarea para ejecutar en segundo plano (write_notification).
• Cualquier secuencia de argumentos que se deben pasar a la función de tarea en orden (email).
• Cualquier argumento por palabra clave que se debe pasar a la función de tarea (message="some notification").

Dependency Injection¶

Usar BackgroundTasks también funciona con el sistema de Dependency Injection, puedes declarar un parámetro de tipo BackgroundTasks en múltiples niveles: en una función de path operation, en una dependencia (dependable), en una sub-dependencia, etc.

FastAPI sabe qué hacer en cada caso y cómo reutilizar el mismo objeto, de modo que todas las tareas en segundo plano se combinan y se ejecutan en segundo plano después:

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 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"}

En este ejemplo, los mensajes serán escritos al archivo log.txt después de que se envíe la response.

Si había una query en la request, se escribirá en el log en una tarea en segundo plano.

Y luego otra tarea en segundo plano generada en la función de path operation escribirá un mensaje usando el parámetro de path email.

Detalles Técnicos¶

La clase BackgroundTasks viene directamente de starlette.background.

Es importada/incluida directamente en FastAPI para que puedas importarla desde fastapi y evitar importar accidentalmente la alternativa BackgroundTask (sin la s al final) desde starlette.background.

Usando solo BackgroundTasks (y no BackgroundTask), es posible usarlo como parámetro de función de path operation y dejar que FastAPI maneje el resto por ti, igual que cuando usas el objeto Request directamente.

Sigue siendo posible usar BackgroundTask solo en FastAPI, pero tienes que crear el objeto en tu código y devolver una Response de Starlette incluyéndolo.

Puedes ver más detalles en la documentación oficial de Starlette para Background Tasks.

Advertencia¶

Si necesitas realizar cálculos pesados en segundo plano y no necesariamente necesitas que se ejecuten en el mismo proceso (por ejemplo, no necesitas compartir memoria, variables, etc), podrías beneficiarte de usar otras herramientas más grandes como Celery.

Tienden a requerir configuraciones más complejas, un gestor de cola de mensajes/trabajos, como RabbitMQ o Redis, pero te permiten ejecutar tareas en segundo plano en múltiples procesos, y especialmente, en múltiples servidores.

Pero si necesitas acceder a variables y objetos de la misma app de FastAPI, o necesitas realizar pequeñas tareas en segundo plano (como enviar una notificación por email), puedes simplemente usar BackgroundTasks.

Resumen¶

Importa y usa BackgroundTasks con parámetros en funciones de path operation y dependencias para añadir tareas en segundo plano.