Ir para o conte√ļdo

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.