Background Tasks¶
Like Lilya, in Esmerald you can define background tasks to run after the returning response.
This can be useful for those operations that need to happen after the request without blocking the client (the client doesn't have to wait to complete) from receiving that same response.
Example:
- Registering a user in the system and send an email confirming the registration.
- Processing a file that can take "some time". Simply return a HTTP 202 and process the file in the background.
How to use¶
As mentioned before, Esmerald uses the background tasks from Lilya and you can pass them in different ways:
Via handlers¶
When via handlers, this means passing a background task via get, put, post, delete or route.
There are also two ways of passing via handlers.
Using a single instance¶
This is probably the most common use case where you simply need to execute one background task upon receiving the request, for example, sending an email notification.
from pydantic import BaseModel
from esmerald import BackgroundTask, JSONResponse, post
class UserIn(BaseModel):
email: str
password: str
async def send_email_notification(message: str):
"""Sends an email notification"""
send_notification(message)
@post(
"/register",
background=BackgroundTask(send_email_notification, message="Account created"),
)
async def create_user(data: UserIn) -> JSONResponse:
JSONResponse({"message": "Created"})
This is of course something quite small as an example but it illustrates how you could use the handlers to pass a background task from there.
Using a list¶
Of course there is also the situation where more than one background task needs to happen.
from datetime import datetime
from pydantic import BaseModel
from esmerald import BackgroundTask, BackgroundTasks, JSONResponse, post
class UserIn(BaseModel):
email: str
password: str
async def send_email_notification(message: str):
"""Sends an email notification"""
send_notification(message)
def write_in_file():
with open("log.txt", mode="w") as log:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
content = f"Notification sent @ {now}"
log.write(content)
@post(
"/register",
background=BackgroundTasks(
tasks=[
BackgroundTask(send_email_notification, message="Account created"),
BackgroundTask(write_in_file),
]
),
)
async def create_user(data: UserIn) -> JSONResponse:
JSONResponse({"message": "Created"})
Via response¶
Adding tasks via response will be probably the way you will be using more often and the reson being is that sometimes you will need some specific information that is only available inside your view.
That is achieved in a similar way as the handlers.
Using a single instance¶
In the same way you created a singe background task for the handlers, in the response works in a similar way.
from typing import Dict
from pydantic import BaseModel
from esmerald import BackgroundTask, Request, Response, post
class UserIn(BaseModel):
email: str
password: str
async def send_email_notification(email: str, message: str):
"""Sends an email notification"""
send_notification(email, message)
@post("/register")
async def create_user(data: UserIn, request: Request) -> Response(Dict[str, str]):
return Response(
{"message": "Email sent"},
background=BackgroundTask(
send_email_notification,
email=request.user.email,
message="Thank you for registering.",
),
)
Using a list¶
The same happens when executing more than one background task and when more than one operation is needed.
from datetime import datetime
from typing import Dict
from pydantic import BaseModel
from esmerald import BackgroundTask, BackgroundTasks, Request, Response, post
class UserIn(BaseModel):
email: str
password: str
async def send_email_notification(email: str, message: str):
"""Sends an email notification"""
send_notification(email, message)
def write_in_file(email: str):
with open("log.txt", mode="w") as log:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
content = f"Notification sent @ {now} to: {email}"
log.write(content)
@post("/register")
async def create_user(data: UserIn, request: Request) -> Response(Dict[str, str]):
return Response(
{"message": "Email sent"},
background=BackgroundTasks(
tasks=[
BackgroundTask(
send_email_notification,
email=request.user.email,
message="Thank you for registering.",
),
BackgroundTask(write_in_file, email=request.user.email),
]
),
)
Using the add_task¶
Another way of adding multiple tasks is by using the add_task function provided by the
BackgroundTasks object.
from datetime import datetime
from typing import Dict
from pydantic import BaseModel
from esmerald import BackgroundTasks, Request, Response, post
class UserIn(BaseModel):
email: str
password: str
async def send_email_notification(email: str, message: str):
"""Sends an email notification"""
send_notification(email, message)
def write_in_file(email: str):
with open("log.txt", mode="w") as log:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
content = f"Notification sent @ {now} to: {email}"
log.write(content)
@post("/register")
async def create_user(data: UserIn, request: Request) -> Response(Dict[str, str]):
background_tasks = BackgroundTasks()
background_tasks.add_task(
send_email_notification, email=request.user.email, message="Thank you for registering."
)
background_tasks.add_task(write_in_file, email=request.user.email)
return Response({"message": "Email sent"}, background=background_tasks)
The .add_task() receives as arguments:
- A task function to be run in the background (send_email_notification and write_in_file).
- Any sequence of arguments that should be passed to the task function in order (email, message).
- Any keyword arguments that should be passed to the task function.
Technical information¶
The class BackgroundTask and BackgroundTasks come directly from lilya.background. This
means you can import directly from Lilya if you want.
Esmerald imports those classes and adds some extra typing information but without affecting the overall functionality of the core.
You can use def or async def functions when declaring those functionalities to be passed to
the BackgroundTask and Esmerald will know how to handle those for you.
For more details about these objects, you can see in Lilya official docs for Background Tasks.
API Reference¶
Learn more about the BackgroundTask by checking the API Reference.