--- /dev/null
+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"}
--- /dev/null
+from fastapi import BackgroundTasks, Depends, FastAPI
+
+app = FastAPI()
+
+
+def write_log(message: str):
+ with open("log.txt", mode="a") as log:
+ log.write(message)
+
+
+def get_query(background_tasks: BackgroundTasks, q: str = None):
+ if q:
+ message = f"found query: {q}\n"
+ background_tasks.add_task(write_log, message)
+ return q
+
+
+@app.post("/send-notification/{email}")
+async def send_notification(
+ email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
+):
+ message = f"message to {email}\n"
+ background_tasks.add_task(write_log, message)
+ return {"message": "Message sent"}
--- /dev/null
+You can define background tasks to be run *after* returning a response.
+
+This is useful for operations that need to happen after a request, but that the client doesn't really have to be waiting for the operation to complete before receiving his response.
+
+This includes, for example:
+
+* Email notifications sent after performing an action:
+ * As connecting to an email server and sending an email tends to be "slow" (several seconds), you can return the response right away and send the email notification in the background.
+* Processing data:
+ * For example, let's say you receive a file that must go through a slow process, you can return a response of "Accepted" (HTTP 202) and process it in the background.
+
+## Using `BackgroundTasks`
+
+First, import `BackgroundTasks` and define a parameter in your *path operation function* with a type declaration of `BackgroundTasks`:
+
+```Python hl_lines="1 13"
+{!./src/background_tasks/tutorial001.py!}
+```
+
+**FastAPI** will create the object of type `BackgroundTasks` for you and pass it as that parameter.
+
+!!! tip
+ You declare a parameter of `BackgroundTasks` and use it in a very similar way as to when <a href="/tutorial/using-request-directly/" target="_blank">using the `Request` directly</a>.
+
+
+## Create a task function
+
+Create a function to be run as the background task.
+
+It is just a standard function that can receive parameters.
+
+It can be an `async def` or normal `def` function, **FastAPI** will know how to handle it correctly.
+
+In this case, the task function will write to a file (simulating sending an email).
+
+And as the write operation doesn't use `async` and `await`, we define the function with normal `def`:
+
+```Python hl_lines="6 7 8 9"
+{!./src/background_tasks/tutorial001.py!}
+```
+
+## Add the background task
+
+Inside of your *path operation function*, pass your task function to the *background tasks* object with the method `.add_task()`:
+
+```Python hl_lines="14"
+{!./src/background_tasks/tutorial001.py!}
+```
+
+`.add_task()` receives as arguments:
+
+* A task function to be run in the background (`write_notification`).
+* Any sequence of arguments that should be passed to the task function in order (`email`).
+* Any keyword arguments that should be passed to the task function (`message="some notification"`).
+
+## Dependency Injection
+
+Using `BackgroundTasks` also works with the dependency injection system, you can declare a parameter of type `BackgroundTasks` at multiple levels: in a *path operation function*, in a dependency (dependable), in a sub-dependency, etc.
+
+**FastAPI** knows what to do in each case and how to re-use the same object, so that all the background tasks are merged together and are run in the background afterwards:
+
+```Python hl_lines="11 14 20 23"
+{!./src/background_tasks/tutorial002.py!}
+```
+
+In this example, the messages will be written to the `log.txt` file *after* the response is sent.
+
+If there was a query in the request, it will be written to the log in a background task.
+
+And then another background task generated at the *path operation function* will write a message using the `email` path parameter.
+
+## Technical Details
+
+The class `BackgroundTasks` comes directly from <a href="https://www.starlette.io/background/" target="_blank">`starlette.background`</a>.
+
+It is imported/included directly into FastAPI so that you can import it from `fastapi` and avoid accidentally importing the alternative `BackgroundTask` (without the `s` at the end) from `starlette.background`.
+
+By only using `BackgroundTasks` (and not `BackgroundTask`), it's then possible to use it as a *path operation function* parameter and have **FastAPI** handle the rest for you, just like when using the `Request` object directly.
+
+It's still possible to use `BackgroundTask` alone in FastAPI, but you have to create the object in your code and return a Starlette `Response` including it.
+
+You can see more details in <a href="https://www.starlette.io/background/" target="_blank">Starlette's official docs for Background Tasks</a>.
+
+## Recap
+
+Import and use `BackgroundTasks` with parameters in *path operation functions* and dependencies to add background tasks.
__version__ = "0.9.1"
+from starlette.background import BackgroundTasks
+
from .applications import FastAPI
from .datastructures import UploadFile
from .exceptions import HTTPException
name: str = None,
call: Callable = None,
request_param_name: str = None,
+ background_tasks_param_name: str = None,
) -> None:
self.path_params = path_params or []
self.query_params = query_params or []
self.dependencies = dependencies or []
self.security_requirements = security_schemes or []
self.request_param_name = request_param_name
+ self.background_tasks_param_name = background_tasks_param_name
self.name = name
self.call = call
from copy import deepcopy
from datetime import date, datetime, time, timedelta
from decimal import Decimal
-from typing import Any, Callable, Dict, List, Mapping, Sequence, Tuple, Type, Union
+from typing import (
+ Any,
+ Callable,
+ Dict,
+ List,
+ Mapping,
+ Optional,
+ Sequence,
+ Tuple,
+ Type,
+ Union,
+)
from uuid import UUID
from fastapi import params
from pydantic.fields import Field, Required, Shape
from pydantic.schema import get_annotation_from_schema
from pydantic.utils import lenient_issubclass
+from starlette.background import BackgroundTasks
from starlette.concurrency import run_in_threadpool
from starlette.datastructures import UploadFile
from starlette.requests import Headers, QueryParams, Request
)
elif lenient_issubclass(param.annotation, Request):
dependant.request_param_name = param_name
+ elif lenient_issubclass(param.annotation, BackgroundTasks):
+ dependant.background_tasks_param_name = param_name
elif not isinstance(param.default, params.Depends):
add_param_to_body_fields(param=param, dependant=dependant)
return dependant
async def solve_dependencies(
- *, request: Request, dependant: Dependant, body: Dict[str, Any] = None
-) -> Tuple[Dict[str, Any], List[ErrorWrapper]]:
+ *,
+ request: Request,
+ dependant: Dependant,
+ body: Dict[str, Any] = None,
+ background_tasks: BackgroundTasks = None,
+) -> Tuple[Dict[str, Any], List[ErrorWrapper], Optional[BackgroundTasks]]:
values: Dict[str, Any] = {}
errors: List[ErrorWrapper] = []
for sub_dependant in dependant.dependencies:
- sub_values, sub_errors = await solve_dependencies(
- request=request, dependant=sub_dependant, body=body
+ sub_values, sub_errors, background_tasks = await solve_dependencies(
+ request=request,
+ dependant=sub_dependant,
+ body=body,
+ background_tasks=background_tasks,
)
if sub_errors:
errors.extend(sub_errors)
errors.extend(body_errors)
if dependant.request_param_name:
values[dependant.request_param_name] = request
- return values, errors
+ if dependant.background_tasks_param_name:
+ if background_tasks is None:
+ background_tasks = BackgroundTasks()
+ values[dependant.background_tasks_param_name] = background_tasks
+ return values, errors, background_tasks
def request_params_to_args(
raise HTTPException(
status_code=400, detail="There was an error parsing the body"
)
- values, errors = await solve_dependencies(
+ values, errors, background_tasks = await solve_dependencies(
request=request, dependant=dependant, body=body
)
if errors:
else:
raw_response = await run_in_threadpool(dependant.call, **values)
if isinstance(raw_response, Response):
+ if raw_response.background is None:
+ raw_response.background = background_tasks
return raw_response
response_data = serialize_response(
field=response_field, response=raw_response
)
- return content_type(content=response_data, status_code=status_code)
+ return content_type(
+ content=response_data,
+ status_code=status_code,
+ background=background_tasks,
+ )
return app
- SQL (Relational) Databases: 'tutorial/sql-databases.md'
- NoSQL (Distributed / Big Data) Databases: 'tutorial/nosql-databases.md'
- Bigger Applications - Multiple Files: 'tutorial/bigger-applications.md'
+ - Background Tasks: 'tutorial/background-tasks.md'
- Sub Applications - Behind a Proxy: 'tutorial/sub-applications-proxy.md'
- Application Configuration: 'tutorial/application-configuration.md'
- GraphQL: 'tutorial/graphql.md'
--- /dev/null
+import os
+from pathlib import Path
+
+from starlette.testclient import TestClient
+
+from background_tasks.tutorial001 import app
+
+client = TestClient(app)
+
+
+def test():
+ log = Path("log.txt")
+ if log.is_file():
+ os.remove(log) # pragma: no cover
+ response = client.post("/send-notification/foo@example.com")
+ assert response.status_code == 200
+ assert response.json() == {"message": "Notification sent in the background"}
+ with open("./log.txt") as f:
+ assert "notification for foo@example.com: some notification" in f.read()
--- /dev/null
+import os
+from pathlib import Path
+
+from starlette.testclient import TestClient
+
+from background_tasks.tutorial002 import app
+
+client = TestClient(app)
+
+
+def test():
+ log = Path("log.txt")
+ if log.is_file():
+ os.remove(log) # pragma: no cover
+ response = client.post("/send-notification/foo@example.com?q=some-query")
+ assert response.status_code == 200
+ assert response.json() == {"message": "Message sent"}
+ with open("./log.txt") as f:
+ assert "found query: some-query\nmessage to foo@example.com" in f.read()
projects / thirdparty / fastapi / fastapi.git / commitdiff
