Background¶
This is the reference for the BackgroundTask
and BackgroundTasks
objects where it contains
all API informationhow to use it.
Like Starlette and any other Starlette based frameworks, 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.
esmerald.BackgroundTask ¶
BackgroundTask(func, *args, **kwargs)
Bases: BackgroundTask
BackgroundTask
as a single instance can be easily achieved.
Example
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"})
PARAMETER | DESCRIPTION |
---|---|
func |
TYPE:
|
*args |
TYPE:
|
**kwargs |
TYPE:
|
Source code in esmerald/backgound.py
45 46 |
|
esmerald.BackgroundTasks ¶
BackgroundTasks(tasks=None)
Bases: BackgroundTasks
Alternatively, the BackgroundTasks
can also be used to be passed
in.
Example
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"})
PARAMETER | DESCRIPTION |
---|---|
tasks |
TYPE:
|
Source code in esmerald/backgound.py
97 98 |
|
add_task ¶
add_task(func, *args, **kwargs)
PARAMETER | DESCRIPTION |
---|---|
func |
TYPE:
|
*args |
TYPE:
|
**kwargs |
TYPE:
|
Source code in .venv/lib/python3.8/site-packages/starlette/background.py
35 36 37 38 39 |
|