Spaces:
Running
Running
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) | |