Tarefas em segundo plano¶

Você pode definir tarefas em segundo plano a serem executadas _ após _ retornar uma resposta.

Isso é útil para operações que precisam acontecer após uma solicitação, mas que o cliente realmente não precisa esperar a operação ser concluída para receber a resposta.

Isso inclui, por exemplo:

Envio de notificações por email após a realização de uma ação:
Como conectar-se a um servidor de e-mail e enviar um e-mail tende a ser "lento" (vários segundos), você pode retornar a resposta imediatamente e enviar a notificação por e-mail em segundo plano.
Processando dados:
Por exemplo, digamos que você receba um arquivo que deve passar por um processo lento, você pode retornar uma resposta de "Aceito" (HTTP 202) e processá-lo em segundo plano.

Usando `BackgroundTasks`¶

Primeiro, importe BackgroundTasks e defina um parâmetro em sua função de operação de caminho com uma declaração de tipo de 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"}

O FastAPI criará o objeto do tipo BackgroundTasks para você e o passará como esse parâmetro.

Criar uma função de tarefa¶

Crie uma função a ser executada como tarefa em segundo plano.

É apenas uma função padrão que pode receber parâmetros.

Pode ser uma função async def ou def normal, o FastAPI saberá como lidar com isso corretamente.

Nesse caso, a função de tarefa gravará em um arquivo (simulando o envio de um e-mail).

E como a operação de gravação não usa async e await, definimos a função com 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"}

Adicionar a tarefa em segundo plano¶

Dentro de sua função de operação de caminho, passe sua função de tarefa para o objeto tarefas em segundo plano com o 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() recebe como argumentos:

Uma função de tarefa a ser executada em segundo plano (write_notification).
Qualquer sequência de argumentos que deve ser passada para a função de tarefa na ordem (email).
Quaisquer argumentos nomeados que devem ser passados para a função de tarefa (mensagem = "alguma notificação").

Injeção de dependência¶

Usar BackgroundTasks também funciona com o sistema de injeção de dependência, você pode declarar um parâmetro do tipo BackgroundTasks em vários níveis: em uma função de operação de caminho, em uma dependência (confiável), em uma subdependência, etc.

O FastAPI sabe o que fazer em cada caso e como reutilizar o mesmo objeto, de forma que todas as tarefas em segundo plano sejam mescladas e executadas em segundo plano posteriormente:

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

Neste exemplo, as mensagens serão gravadas no arquivo log.txt após o envio da resposta.

Se houver uma consulta na solicitação, ela será gravada no log em uma tarefa em segundo plano.

E então outra tarefa em segundo plano gerada na função de operação de caminho escreverá uma mensagem usando o parâmetro de caminho email.

Detalhes técnicos¶

A classe BackgroundTasks vem diretamente de starlette.background.

Ela é importada/incluída diretamente no FastAPI para que você possa importá-la do fastapi e evitar a importação acidental da alternativa BackgroundTask (sem o s no final) de starlette.background.

Usando apenas BackgroundTasks (e não BackgroundTask), é então possível usá-la como um parâmetro de função de operação de caminho e deixar o FastAPI cuidar do resto para você, assim como ao usar o objeto Request diretamente.

Ainda é possível usar BackgroundTask sozinho no FastAPI, mas você deve criar o objeto em seu código e retornar uma Starlette Response incluindo-o.

Você pode ver mais detalhes na documentação oficiais da Starlette para tarefas em segundo plano .

Ressalva¶

Se você precisa realizar cálculos pesados em segundo plano e não necessariamente precisa que seja executado pelo mesmo processo (por exemplo, você não precisa compartilhar memória, variáveis, etc), você pode se beneficiar do uso de outras ferramentas maiores, como Celery .

Eles tendem a exigir configurações mais complexas, um gerenciador de fila de mensagens/tarefas, como RabbitMQ ou Redis, mas permitem que você execute tarefas em segundo plano em vários processos e, especialmente, em vários servidores.

Para ver um exemplo, verifique os Geradores de projeto, todos incluem celery já configurado.

Mas se você precisa acessar variáveis e objetos do mesmo aplicativo FastAPI, ou precisa realizar pequenas tarefas em segundo plano (como enviar uma notificação por e-mail), você pode simplesmente usar BackgroundTasks.

Recapitulando¶

Importe e use BackgroundTasks com parâmetros em funções de operação de caminho e dependências para adicionar tarefas em segundo plano.