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"}
🤓 Other versions and variants
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"}
Tip
Prefer to use the Annotated version if possible.
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"}
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.
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.