|
from typing import Any, Callable |
|
|
|
from starlette.background import BackgroundTasks as StarletteBackgroundTasks |
|
from typing_extensions import Annotated, Doc, ParamSpec |
|
|
|
P = ParamSpec("P") |
|
|
|
|
|
class BackgroundTasks(StarletteBackgroundTasks): |
|
""" |
|
A collection of background tasks that will be called after a response has been |
|
sent to the client. |
|
|
|
Read more about it in the |
|
[FastAPI docs for Background Tasks](https://fastapi.tiangolo.com/tutorial/background-tasks/). |
|
|
|
## Example |
|
|
|
```python |
|
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"} |
|
``` |
|
""" |
|
|
|
def add_task( |
|
self, |
|
func: Annotated[ |
|
Callable[P, Any], |
|
Doc( |
|
""" |
|
The function to call after the response is sent. |
|
|
|
It can be a regular `def` function or an `async def` function. |
|
""" |
|
), |
|
], |
|
*args: P.args, |
|
**kwargs: P.kwargs, |
|
) -> None: |
|
""" |
|
Add a function to be called in the background after the response is sent. |
|
|
|
Read more about it in the |
|
[FastAPI docs for Background Tasks](https://fastapi.tiangolo.com/tutorial/background-tasks/). |
|
""" |
|
return super().add_task(func, *args, **kwargs) |
|
|