background.py 1.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  1. from collections.abc import Callable
  2. from typing import Annotated, Any
  3. from annotated_doc import Doc
  4. from starlette.background import BackgroundTasks as StarletteBackgroundTasks
  5. from typing_extensions import ParamSpec
  6. P = ParamSpec("P")
  7. class BackgroundTasks(StarletteBackgroundTasks):
  8. """
  9. A collection of background tasks that will be called after a response has been
  10. sent to the client.
  11. Read more about it in the
  12. [FastAPI docs for Background Tasks](https://fastapi.tiangolo.com/tutorial/background-tasks/).
  13. ## Example
  14. ```python
  15. from fastapi import BackgroundTasks, FastAPI
  16. app = FastAPI()
  17. def write_notification(email: str, message=""):
  18. with open("log.txt", mode="w") as email_file:
  19. content = f"notification for {email}: {message}"
  20. email_file.write(content)
  21. @app.post("/send-notification/{email}")
  22. async def send_notification(email: str, background_tasks: BackgroundTasks):
  23. background_tasks.add_task(write_notification, email, message="some notification")
  24. return {"message": "Notification sent in the background"}
  25. ```
  26. """
  27. def add_task(
  28. self,
  29. func: Annotated[
  30. Callable[P, Any],
  31. Doc(
  32. """
  33. The function to call after the response is sent.
  34. It can be a regular `def` function or an `async def` function.
  35. """
  36. ),
  37. ],
  38. *args: P.args,
  39. **kwargs: P.kwargs,
  40. ) -> None:
  41. """
  42. Add a function to be called in the background after the response is sent.
  43. Read more about it in the
  44. [FastAPI docs for Background Tasks](https://fastapi.tiangolo.com/tutorial/background-tasks/).
  45. """
  46. return super().add_task(func, *args, **kwargs)