For example, you can add an additional media type of `image/png`, declaring that your *path operation* can return a JSON object (with media type `application/json`) or a PNG image:
-{* ../../docs_src/additional_responses/tutorial002.py hl[19:24,28] *}
+{* ../../docs_src/additional_responses/tutorial002_py310.py hl[17:22,26] *}
/// note
For example:
-{* ../../docs_src/additional_responses/tutorial004.py hl[13:17,26] *}
+{* ../../docs_src/additional_responses/tutorial004_py310.py hl[11:15,24] *}
## More information about OpenAPI responses { #more-information-about-openapi-responses }
But FastAPI also supports using <a href="https://docs.python.org/3/library/dataclasses.html" class="external-link" target="_blank">`dataclasses`</a> the same way:
-{* ../../docs_src/dataclasses/tutorial001.py hl[1,7:12,19:20] *}
+{* ../../docs_src/dataclasses/tutorial001_py310.py hl[1,6:11,18:19] *}
This is still supported thanks to **Pydantic**, as it has <a href="https://docs.pydantic.dev/latest/concepts/dataclasses/#use-of-stdlib-dataclasses-with-basemodel" class="external-link" target="_blank">internal support for `dataclasses`</a>.
You can also use `dataclasses` in the `response_model` parameter:
-{* ../../docs_src/dataclasses/tutorial002.py hl[1,7:13,19] *}
+{* ../../docs_src/dataclasses/tutorial002_py310.py hl[1,6:12,18] *}
The dataclass will be automatically converted to a Pydantic dataclass.
In that case, you can simply swap the standard `dataclasses` with `pydantic.dataclasses`, which is a drop-in replacement:
-{* ../../docs_src/dataclasses/tutorial003.py hl[1,5,8:11,14:17,23:25,28] *}
+{* ../../docs_src/dataclasses/tutorial003_py310.py hl[1,4,7:10,13:16,22:24,27] *}
1. We still import `field` from standard `dataclasses`.
This part is pretty normal, most of the code is probably already familiar to you:
-{* ../../docs_src/openapi_callbacks/tutorial001.py hl[9:13,36:53] *}
+{* ../../docs_src/openapi_callbacks/tutorial001_py310.py hl[7:11,34:51] *}
/// tip
First create a new `APIRouter` that will contain one or more callbacks.
-{* ../../docs_src/openapi_callbacks/tutorial001.py hl[3,25] *}
+{* ../../docs_src/openapi_callbacks/tutorial001_py310.py hl[1,23] *}
### Create the callback *path operation* { #create-the-callback-path-operation }
* It should probably have a declaration of the body it should receive, e.g. `body: InvoiceEvent`.
* And it could also have a declaration of the response it should return, e.g. `response_model=InvoiceEventReceived`.
-{* ../../docs_src/openapi_callbacks/tutorial001.py hl[16:18,21:22,28:32] *}
+{* ../../docs_src/openapi_callbacks/tutorial001_py310.py hl[14:16,19:20,26:30] *}
There are 2 main differences from a normal *path operation*:
Now use the parameter `callbacks` in *your API's path operation decorator* to pass the attribute `.routes` (that's actually just a `list` of routes/*path operations*) from that callback router:
-{* ../../docs_src/openapi_callbacks/tutorial001.py hl[35] *}
+{* ../../docs_src/openapi_callbacks/tutorial001_py310.py hl[33] *}
/// tip
It won't show up in the documentation, but other tools (such as Sphinx) will be able to use the rest.
-{* ../../docs_src/path_operation_advanced_configuration/tutorial004.py hl[19:29] *}
+{* ../../docs_src/path_operation_advanced_configuration/tutorial004_py310.py hl[17:27] *}
## Additional Responses { #additional-responses }
//// tab | Pydantic v2
-{* ../../docs_src/path_operation_advanced_configuration/tutorial007.py hl[17:22, 24] *}
+{* ../../docs_src/path_operation_advanced_configuration/tutorial007_py39.py hl[15:20, 22] *}
////
//// tab | Pydantic v1
-{* ../../docs_src/path_operation_advanced_configuration/tutorial007_pv1.py hl[17:22, 24] *}
+{* ../../docs_src/path_operation_advanced_configuration/tutorial007_pv1_py39.py hl[15:20, 22] *}
////
//// tab | Pydantic v2
-{* ../../docs_src/path_operation_advanced_configuration/tutorial007.py hl[26:33] *}
+{* ../../docs_src/path_operation_advanced_configuration/tutorial007_py39.py hl[24:31] *}
////
//// tab | Pydantic v1
-{* ../../docs_src/path_operation_advanced_configuration/tutorial007_pv1.py hl[26:33] *}
+{* ../../docs_src/path_operation_advanced_configuration/tutorial007_pv1_py39.py hl[24:31] *}
////
For those cases, you can use the `jsonable_encoder` to convert your data before passing it to a response:
-{* ../../docs_src/response_directly/tutorial001.py hl[6:7,21:22] *}
+{* ../../docs_src/response_directly/tutorial001_py310.py hl[5:6,20:21] *}
/// note | Technical Details
Coming from the previous example, your `config.py` file could look like:
-{* ../../docs_src/settings/app02/config.py hl[10] *}
+{* ../../docs_src/settings/app02_an_py39/config.py hl[10] *}
Notice that now we don't create a default instance `settings = Settings()`.
Then it would be very easy to provide a different settings object during testing by creating a dependency override for `get_settings`:
-{* ../../docs_src/settings/app02/test_main.py hl[9:10,13,21] *}
+{* ../../docs_src/settings/app02_an_py39/test_main.py hl[9:10,13,21] *}
In the dependency override we set a new value for the `admin_email` when creating the new `Settings` object, and then we return that new object.
//// tab | Pydantic v2
-{* ../../docs_src/settings/app03_an/config.py hl[9] *}
+{* ../../docs_src/settings/app03_an_py39/config.py hl[9] *}
/// tip
//// tab | Pydantic v1
-{* ../../docs_src/settings/app03_an/config_pv1.py hl[9:10] *}
+{* ../../docs_src/settings/app03_an_py39/config_pv1.py hl[9:10] *}
/// tip
It includes these default configurations:
-{* ../../fastapi/openapi/docs.py ln[8:23] hl[17:23] *}
+{* ../../fastapi/openapi/docs.py ln[9:24] hl[18:24] *}
You can override any of them by setting a different value in the argument `swagger_ui_parameters`.
That way, the same route class can handle gzip compressed or uncompressed requests.
-{* ../../docs_src/custom_request_and_route/tutorial001.py hl[8:15] *}
+{* ../../docs_src/custom_request_and_route/tutorial001_an_py310.py hl[9:16] *}
### Create a custom `GzipRoute` class { #create-a-custom-gziproute-class }
Here we use it to create a `GzipRequest` from the original request.
-{* ../../docs_src/custom_request_and_route/tutorial001.py hl[18:26] *}
+{* ../../docs_src/custom_request_and_route/tutorial001_an_py310.py hl[19:27] *}
/// note | Technical Details
All we need to do is handle the request inside a `try`/`except` block:
-{* ../../docs_src/custom_request_and_route/tutorial002.py hl[13,15] *}
+{* ../../docs_src/custom_request_and_route/tutorial002_an_py310.py hl[14,16] *}
If an exception occurs, the`Request` instance will still be in scope, so we can read and make use of the request body when handling the error:
-{* ../../docs_src/custom_request_and_route/tutorial002.py hl[16:18] *}
+{* ../../docs_src/custom_request_and_route/tutorial002_an_py310.py hl[17:19] *}
## Custom `APIRoute` class in a router { #custom-apiroute-class-in-a-router }
You can also set the `route_class` parameter of an `APIRouter`:
-{* ../../docs_src/custom_request_and_route/tutorial003.py hl[26] *}
+{* ../../docs_src/custom_request_and_route/tutorial003_py310.py hl[26] *}
In this example, the *path operations* under the `router` will use the custom `TimedRoute` class, and will have an extra `X-Response-Time` header in the response with the time it took to generate the response:
-{* ../../docs_src/custom_request_and_route/tutorial003.py hl[13:20] *}
+{* ../../docs_src/custom_request_and_route/tutorial003_py310.py hl[13:20] *}
You import it and create an "instance" the same way you would with the class `FastAPI`:
-```Python hl_lines="1 3" title="app/routers/users.py"
-{!../../docs_src/bigger_applications/app/routers/users.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/routers/users.py hl[1,3] title["app/routers/users.py"] *}
### *Path operations* with `APIRouter` { #path-operations-with-apirouter }
Use it the same way you would use the `FastAPI` class:
-```Python hl_lines="6 11 16" title="app/routers/users.py"
-{!../../docs_src/bigger_applications/app/routers/users.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/routers/users.py hl[6,11,16] title["app/routers/users.py"] *}
You can think of `APIRouter` as a "mini `FastAPI`" class.
We will now use a simple dependency to read a custom `X-Token` header:
-//// tab | Python 3.9+
-
-```Python hl_lines="3 6-8" title="app/dependencies.py"
-{!> ../../docs_src/bigger_applications/app_an_py39/dependencies.py!}
-```
-
-////
-
-//// tab | Python 3.8+
-
-```Python hl_lines="1 5-7" title="app/dependencies.py"
-{!> ../../docs_src/bigger_applications/app_an/dependencies.py!}
-```
-
-////
-
-//// tab | Python 3.8+ non-Annotated
-
-/// tip
-
-Prefer to use the `Annotated` version if possible.
-
-///
-
-```Python hl_lines="1 4-6" title="app/dependencies.py"
-{!> ../../docs_src/bigger_applications/app/dependencies.py!}
-```
-
-////
+{* ../../docs_src/bigger_applications/app_an_py39/dependencies.py hl[3,6:8] title["app/dependencies.py"] *}
/// tip
So, instead of adding all that to each *path operation*, we can add it to the `APIRouter`.
-```Python hl_lines="5-10 16 21" title="app/routers/items.py"
-{!../../docs_src/bigger_applications/app/routers/items.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/routers/items.py hl[5:10,16,21] title["app/routers/items.py"] *}
As the path of each *path operation* has to start with `/`, like in:
So we use a relative import with `..` for the dependencies:
-```Python hl_lines="3" title="app/routers/items.py"
-{!../../docs_src/bigger_applications/app/routers/items.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/routers/items.py hl[3] title["app/routers/items.py"] *}
#### How relative imports work { #how-relative-imports-work }
But we can still add _more_ `tags` that will be applied to a specific *path operation*, and also some extra `responses` specific to that *path operation*:
-```Python hl_lines="30-31" title="app/routers/items.py"
-{!../../docs_src/bigger_applications/app/routers/items.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/routers/items.py hl[30:31] title["app/routers/items.py"] *}
/// tip
And we can even declare [global dependencies](dependencies/global-dependencies.md){.internal-link target=_blank} that will be combined with the dependencies for each `APIRouter`:
-```Python hl_lines="1 3 7" title="app/main.py"
-{!../../docs_src/bigger_applications/app/main.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/main.py hl[1,3,7] title["app/main.py"] *}
### Import the `APIRouter` { #import-the-apirouter }
Now we import the other submodules that have `APIRouter`s:
-```Python hl_lines="4-5" title="app/main.py"
-{!../../docs_src/bigger_applications/app/main.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/main.py hl[4:5] title["app/main.py"] *}
As the files `app/routers/users.py` and `app/routers/items.py` are submodules that are part of the same Python package `app`, we can use a single dot `.` to import them using "relative imports".
So, to be able to use both of them in the same file, we import the submodules directly:
-```Python hl_lines="5" title="app/main.py"
-{!../../docs_src/bigger_applications/app/main.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/main.py hl[5] title["app/main.py"] *}
### Include the `APIRouter`s for `users` and `items` { #include-the-apirouters-for-users-and-items }
Now, let's include the `router`s from the submodules `users` and `items`:
-```Python hl_lines="10-11" title="app/main.py"
-{!../../docs_src/bigger_applications/app/main.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/main.py hl[10:11] title["app/main.py"] *}
/// info
For this example it will be super simple. But let's say that because it is shared with other projects in the organization, we cannot modify it and add a `prefix`, `dependencies`, `tags`, etc. directly to the `APIRouter`:
-```Python hl_lines="3" title="app/internal/admin.py"
-{!../../docs_src/bigger_applications/app/internal/admin.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/internal/admin.py hl[3] title["app/internal/admin.py"] *}
But we still want to set a custom `prefix` when including the `APIRouter` so that all its *path operations* start with `/admin`, we want to secure it with the `dependencies` we already have for this project, and we want to include `tags` and `responses`.
We can declare all that without having to modify the original `APIRouter` by passing those parameters to `app.include_router()`:
-```Python hl_lines="14-17" title="app/main.py"
-{!../../docs_src/bigger_applications/app/main.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/main.py hl[14:17] title["app/main.py"] *}
That way, the original `APIRouter` will stay unmodified, so we can still share that same `app/internal/admin.py` file with other projects in the organization.
Here we do it... just to show that we can 🤷:
-```Python hl_lines="21-23" title="app/main.py"
-{!../../docs_src/bigger_applications/app/main.py!}
-```
+{* ../../docs_src/bigger_applications/app_an_py39/main.py hl[21:23] title["app/main.py"] *}
and it will work correctly, together with all the other *path operations* added with `app.include_router()`.
You can use Pydantic's model configuration to `forbid` any `extra` fields:
-{* ../../docs_src/cookie_param_models/tutorial002_an_py39.py hl[10] *}
+{* ../../docs_src/cookie_param_models/tutorial002_an_py310.py hl[10] *}
If a client tries to send some **extra cookies**, they will receive an **error** response.
Both *path operations* require an `X-Token` header.
-//// tab | Python 3.10+
-
-```Python
-{!> ../../docs_src/app_testing/app_b_an_py310/main.py!}
-```
-
-////
-
-//// tab | Python 3.9+
-
-```Python
-{!> ../../docs_src/app_testing/app_b_an_py39/main.py!}
-```
-
-////
-
-//// tab | Python 3.8+
-
-```Python
-{!> ../../docs_src/app_testing/app_b_an/main.py!}
-```
-
-////
-
-//// tab | Python 3.10+ non-Annotated
-
-/// tip
-
-Prefer to use the `Annotated` version if possible.
-
-///
-
-```Python
-{!> ../../docs_src/app_testing/app_b_py310/main.py!}
-```
-
-////
-
-//// tab | Python 3.8+ non-Annotated
-
-/// tip
-
-Prefer to use the `Annotated` version if possible.
-
-///
-
-```Python
-{!> ../../docs_src/app_testing/app_b/main.py!}
-```
-
-////
+{* ../../docs_src/app_testing/app_b_an_py310/main.py *}
### Extended testing file { #extended-testing-file }
You could then update `test_main.py` with the extended tests:
-{* ../../docs_src/app_testing/app_b/test_main.py *}
+{* ../../docs_src/app_testing/app_b_an_py310/test_main.py *}
Whenever you need the client to pass information in the request and you don't know how to, you can search (Google) how to do it in `httpx`, or even how to do it with `requests`, as HTTPX's design is based on Requests' design.
--- /dev/null
+from fastapi import FastAPI
+from fastapi.responses import FileResponse
+from pydantic import BaseModel
+
+
+class Item(BaseModel):
+ id: str
+ value: str
+
+
+app = FastAPI()
+
+
+@app.get(
+ "/items/{item_id}",
+ response_model=Item,
+ responses={
+ 200: {
+ "content": {"image/png": {}},
+ "description": "Return the JSON item or an image.",
+ }
+ },
+)
+async def read_item(item_id: str, img: bool | None = None):
+ if img:
+ return FileResponse("image.png", media_type="image/png")
+ else:
+ return {"id": "foo", "value": "there goes my hero"}
--- /dev/null
+from fastapi import FastAPI
+from fastapi.responses import FileResponse
+from pydantic import BaseModel
+
+
+class Item(BaseModel):
+ id: str
+ value: str
+
+
+responses = {
+ 404: {"description": "Item not found"},
+ 302: {"description": "The item was moved"},
+ 403: {"description": "Not enough privileges"},
+}
+
+
+app = FastAPI()
+
+
+@app.get(
+ "/items/{item_id}",
+ response_model=Item,
+ responses={**responses, 200: {"content": {"image/png": {}}}},
+)
+async def read_item(item_id: str, img: bool | None = None):
+ if img:
+ return FileResponse("image.png", media_type="image/png")
+ else:
+ return {"id": "foo", "value": "there goes my hero"}
--- /dev/null
+import gzip
+from typing import Callable, List
+
+from fastapi import Body, FastAPI, Request, Response
+from fastapi.routing import APIRoute
+from typing_extensions import Annotated
+
+
+class GzipRequest(Request):
+ async def body(self) -> bytes:
+ if not hasattr(self, "_body"):
+ body = await super().body()
+ if "gzip" in self.headers.getlist("Content-Encoding"):
+ body = gzip.decompress(body)
+ self._body = body
+ return self._body
+
+
+class GzipRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ request = GzipRequest(request.scope, request.receive)
+ return await original_route_handler(request)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = GzipRoute
+
+
+@app.post("/sum")
+async def sum_numbers(numbers: Annotated[List[int], Body()]):
+ return {"sum": sum(numbers)}
--- /dev/null
+import gzip
+from collections.abc import Callable
+from typing import Annotated
+
+from fastapi import Body, FastAPI, Request, Response
+from fastapi.routing import APIRoute
+
+
+class GzipRequest(Request):
+ async def body(self) -> bytes:
+ if not hasattr(self, "_body"):
+ body = await super().body()
+ if "gzip" in self.headers.getlist("Content-Encoding"):
+ body = gzip.decompress(body)
+ self._body = body
+ return self._body
+
+
+class GzipRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ request = GzipRequest(request.scope, request.receive)
+ return await original_route_handler(request)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = GzipRoute
+
+
+@app.post("/sum")
+async def sum_numbers(numbers: Annotated[list[int], Body()]):
+ return {"sum": sum(numbers)}
--- /dev/null
+import gzip
+from typing import Annotated, Callable
+
+from fastapi import Body, FastAPI, Request, Response
+from fastapi.routing import APIRoute
+
+
+class GzipRequest(Request):
+ async def body(self) -> bytes:
+ if not hasattr(self, "_body"):
+ body = await super().body()
+ if "gzip" in self.headers.getlist("Content-Encoding"):
+ body = gzip.decompress(body)
+ self._body = body
+ return self._body
+
+
+class GzipRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ request = GzipRequest(request.scope, request.receive)
+ return await original_route_handler(request)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = GzipRoute
+
+
+@app.post("/sum")
+async def sum_numbers(numbers: Annotated[list[int], Body()]):
+ return {"sum": sum(numbers)}
--- /dev/null
+import gzip
+from collections.abc import Callable
+
+from fastapi import Body, FastAPI, Request, Response
+from fastapi.routing import APIRoute
+
+
+class GzipRequest(Request):
+ async def body(self) -> bytes:
+ if not hasattr(self, "_body"):
+ body = await super().body()
+ if "gzip" in self.headers.getlist("Content-Encoding"):
+ body = gzip.decompress(body)
+ self._body = body
+ return self._body
+
+
+class GzipRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ request = GzipRequest(request.scope, request.receive)
+ return await original_route_handler(request)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = GzipRoute
+
+
+@app.post("/sum")
+async def sum_numbers(numbers: list[int] = Body()):
+ return {"sum": sum(numbers)}
--- /dev/null
+import gzip
+from typing import Callable
+
+from fastapi import Body, FastAPI, Request, Response
+from fastapi.routing import APIRoute
+
+
+class GzipRequest(Request):
+ async def body(self) -> bytes:
+ if not hasattr(self, "_body"):
+ body = await super().body()
+ if "gzip" in self.headers.getlist("Content-Encoding"):
+ body = gzip.decompress(body)
+ self._body = body
+ return self._body
+
+
+class GzipRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ request = GzipRequest(request.scope, request.receive)
+ return await original_route_handler(request)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = GzipRoute
+
+
+@app.post("/sum")
+async def sum_numbers(numbers: list[int] = Body()):
+ return {"sum": sum(numbers)}
--- /dev/null
+from typing import Callable, List
+
+from fastapi import Body, FastAPI, HTTPException, Request, Response
+from fastapi.exceptions import RequestValidationError
+from fastapi.routing import APIRoute
+from typing_extensions import Annotated
+
+
+class ValidationErrorLoggingRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ try:
+ return await original_route_handler(request)
+ except RequestValidationError as exc:
+ body = await request.body()
+ detail = {"errors": exc.errors(), "body": body.decode()}
+ raise HTTPException(status_code=422, detail=detail)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = ValidationErrorLoggingRoute
+
+
+@app.post("/")
+async def sum_numbers(numbers: Annotated[List[int], Body()]):
+ return sum(numbers)
--- /dev/null
+from collections.abc import Callable
+from typing import Annotated
+
+from fastapi import Body, FastAPI, HTTPException, Request, Response
+from fastapi.exceptions import RequestValidationError
+from fastapi.routing import APIRoute
+
+
+class ValidationErrorLoggingRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ try:
+ return await original_route_handler(request)
+ except RequestValidationError as exc:
+ body = await request.body()
+ detail = {"errors": exc.errors(), "body": body.decode()}
+ raise HTTPException(status_code=422, detail=detail)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = ValidationErrorLoggingRoute
+
+
+@app.post("/")
+async def sum_numbers(numbers: Annotated[list[int], Body()]):
+ return sum(numbers)
--- /dev/null
+from typing import Annotated, Callable
+
+from fastapi import Body, FastAPI, HTTPException, Request, Response
+from fastapi.exceptions import RequestValidationError
+from fastapi.routing import APIRoute
+
+
+class ValidationErrorLoggingRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ try:
+ return await original_route_handler(request)
+ except RequestValidationError as exc:
+ body = await request.body()
+ detail = {"errors": exc.errors(), "body": body.decode()}
+ raise HTTPException(status_code=422, detail=detail)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = ValidationErrorLoggingRoute
+
+
+@app.post("/")
+async def sum_numbers(numbers: Annotated[list[int], Body()]):
+ return sum(numbers)
--- /dev/null
+from collections.abc import Callable
+
+from fastapi import Body, FastAPI, HTTPException, Request, Response
+from fastapi.exceptions import RequestValidationError
+from fastapi.routing import APIRoute
+
+
+class ValidationErrorLoggingRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ try:
+ return await original_route_handler(request)
+ except RequestValidationError as exc:
+ body = await request.body()
+ detail = {"errors": exc.errors(), "body": body.decode()}
+ raise HTTPException(status_code=422, detail=detail)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = ValidationErrorLoggingRoute
+
+
+@app.post("/")
+async def sum_numbers(numbers: list[int] = Body()):
+ return sum(numbers)
--- /dev/null
+from typing import Callable
+
+from fastapi import Body, FastAPI, HTTPException, Request, Response
+from fastapi.exceptions import RequestValidationError
+from fastapi.routing import APIRoute
+
+
+class ValidationErrorLoggingRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ try:
+ return await original_route_handler(request)
+ except RequestValidationError as exc:
+ body = await request.body()
+ detail = {"errors": exc.errors(), "body": body.decode()}
+ raise HTTPException(status_code=422, detail=detail)
+
+ return custom_route_handler
+
+
+app = FastAPI()
+app.router.route_class = ValidationErrorLoggingRoute
+
+
+@app.post("/")
+async def sum_numbers(numbers: list[int] = Body()):
+ return sum(numbers)
--- /dev/null
+import time
+from collections.abc import Callable
+
+from fastapi import APIRouter, FastAPI, Request, Response
+from fastapi.routing import APIRoute
+
+
+class TimedRoute(APIRoute):
+ def get_route_handler(self) -> Callable:
+ original_route_handler = super().get_route_handler()
+
+ async def custom_route_handler(request: Request) -> Response:
+ before = time.time()
+ response: Response = await original_route_handler(request)
+ duration = time.time() - before
+ response.headers["X-Response-Time"] = str(duration)
+ print(f"route duration: {duration}")
+ print(f"route response: {response}")
+ print(f"route response headers: {response.headers}")
+ return response
+
+ return custom_route_handler
+
+
+app = FastAPI()
+router = APIRouter(route_class=TimedRoute)
+
+
+@app.get("/")
+async def not_timed():
+ return {"message": "Not timed"}
+
+
+@router.get("/timed")
+async def timed():
+ return {"message": "It's the time of my life"}
+
+
+app.include_router(router)
--- /dev/null
+from dataclasses import dataclass
+
+from fastapi import FastAPI
+
+
+@dataclass
+class Item:
+ name: str
+ price: float
+ description: str | None = None
+ tax: float | None = None
+
+
+app = FastAPI()
+
+
+@app.post("/items/")
+async def create_item(item: Item):
+ return item
--- /dev/null
+from dataclasses import dataclass, field
+
+from fastapi import FastAPI
+
+
+@dataclass
+class Item:
+ name: str
+ price: float
+ tags: list[str] = field(default_factory=list)
+ description: str | None = None
+ tax: float | None = None
+
+
+app = FastAPI()
+
+
+@app.get("/items/next", response_model=Item)
+async def read_next_item():
+ return {
+ "name": "Island In The Moon",
+ "price": 12.99,
+ "description": "A place to be playin' and havin' fun",
+ "tags": ["breater"],
+ }
--- /dev/null
+from dataclasses import dataclass, field
+from typing import Union
+
+from fastapi import FastAPI
+
+
+@dataclass
+class Item:
+ name: str
+ price: float
+ tags: list[str] = field(default_factory=list)
+ description: Union[str, None] = None
+ tax: Union[float, None] = None
+
+
+app = FastAPI()
+
+
+@app.get("/items/next", response_model=Item)
+async def read_next_item():
+ return {
+ "name": "Island In The Moon",
+ "price": 12.99,
+ "description": "A place to be playin' and havin' fun",
+ "tags": ["breater"],
+ }
--- /dev/null
+from dataclasses import field # (1)
+
+from fastapi import FastAPI
+from pydantic.dataclasses import dataclass # (2)
+
+
+@dataclass
+class Item:
+ name: str
+ description: str | None = None
+
+
+@dataclass
+class Author:
+ name: str
+ items: list[Item] = field(default_factory=list) # (3)
+
+
+app = FastAPI()
+
+
+@app.post("/authors/{author_id}/items/", response_model=Author) # (4)
+async def create_author_items(author_id: str, items: list[Item]): # (5)
+ return {"name": author_id, "items": items} # (6)
+
+
+@app.get("/authors/", response_model=list[Author]) # (7)
+def get_authors(): # (8)
+ return [ # (9)
+ {
+ "name": "Breaters",
+ "items": [
+ {
+ "name": "Island In The Moon",
+ "description": "A place to be playin' and havin' fun",
+ },
+ {"name": "Holy Buddies"},
+ ],
+ },
+ {
+ "name": "System of an Up",
+ "items": [
+ {
+ "name": "Salt",
+ "description": "The kombucha mushroom people's favorite",
+ },
+ {"name": "Pad Thai"},
+ {
+ "name": "Lonely Night",
+ "description": "The mostests lonliest nightiest of allest",
+ },
+ ],
+ },
+ ]
--- /dev/null
+from dataclasses import field # (1)
+from typing import Union
+
+from fastapi import FastAPI
+from pydantic.dataclasses import dataclass # (2)
+
+
+@dataclass
+class Item:
+ name: str
+ description: Union[str, None] = None
+
+
+@dataclass
+class Author:
+ name: str
+ items: list[Item] = field(default_factory=list) # (3)
+
+
+app = FastAPI()
+
+
+@app.post("/authors/{author_id}/items/", response_model=Author) # (4)
+async def create_author_items(author_id: str, items: list[Item]): # (5)
+ return {"name": author_id, "items": items} # (6)
+
+
+@app.get("/authors/", response_model=list[Author]) # (7)
+def get_authors(): # (8)
+ return [ # (9)
+ {
+ "name": "Breaters",
+ "items": [
+ {
+ "name": "Island In The Moon",
+ "description": "A place to be playin' and havin' fun",
+ },
+ {"name": "Holy Buddies"},
+ ],
+ },
+ {
+ "name": "System of an Up",
+ "items": [
+ {
+ "name": "Salt",
+ "description": "The kombucha mushroom people's favorite",
+ },
+ {"name": "Pad Thai"},
+ {
+ "name": "Lonely Night",
+ "description": "The mostests lonliest nightiest of allest",
+ },
+ ],
+ },
+ ]
--- /dev/null
+from fastapi import APIRouter, FastAPI
+from pydantic import BaseModel, HttpUrl
+
+app = FastAPI()
+
+
+class Invoice(BaseModel):
+ id: str
+ title: str | None = None
+ customer: str
+ total: float
+
+
+class InvoiceEvent(BaseModel):
+ description: str
+ paid: bool
+
+
+class InvoiceEventReceived(BaseModel):
+ ok: bool
+
+
+invoices_callback_router = APIRouter()
+
+
+@invoices_callback_router.post(
+ "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived
+)
+def invoice_notification(body: InvoiceEvent):
+ pass
+
+
+@app.post("/invoices/", callbacks=invoices_callback_router.routes)
+def create_invoice(invoice: Invoice, callback_url: HttpUrl | None = None):
+ """
+ Create an invoice.
+
+ This will (let's imagine) let the API user (some external developer) create an
+ invoice.
+
+ And this path operation will:
+
+ * Send the invoice to the client.
+ * Collect the money from the client.
+ * Send a notification back to the API user (the external developer), as a callback.
+ * At this point is that the API will somehow send a POST request to the
+ external API with the notification of the invoice event
+ (e.g. "payment successful").
+ """
+ # Send the invoice, collect the money, send the notification (the callback)
+ return {"msg": "Invoice received"}
--- /dev/null
+from fastapi import FastAPI
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class Item(BaseModel):
+ name: str
+ description: str | None = None
+ price: float
+ tax: float | None = None
+ tags: set[str] = set()
+
+
+@app.post("/items/", response_model=Item, summary="Create an item")
+async def create_item(item: Item):
+ """
+ Create an item with all the information:
+
+ - **name**: each item must have a name
+ - **description**: a long description
+ - **price**: required
+ - **tax**: if the item doesn't have tax, you can omit this
+ - **tags**: a set of unique tag strings for this item
+ \f
+ :param item: User input.
+ """
+ return item
--- /dev/null
+from typing import Union
+
+from fastapi import FastAPI
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class Item(BaseModel):
+ name: str
+ description: Union[str, None] = None
+ price: float
+ tax: Union[float, None] = None
+ tags: set[str] = set()
+
+
+@app.post("/items/", response_model=Item, summary="Create an item")
+async def create_item(item: Item):
+ """
+ Create an item with all the information:
+
+ - **name**: each item must have a name
+ - **description**: a long description
+ - **price**: required
+ - **tax**: if the item doesn't have tax, you can omit this
+ - **tags**: a set of unique tag strings for this item
+ \f
+ :param item: User input.
+ """
+ return item
--- /dev/null
+import yaml
+from fastapi import FastAPI, HTTPException, Request
+from pydantic import BaseModel, ValidationError
+
+app = FastAPI()
+
+
+class Item(BaseModel):
+ name: str
+ tags: list[str]
+
+
+@app.post(
+ "/items/",
+ openapi_extra={
+ "requestBody": {
+ "content": {"application/x-yaml": {"schema": Item.schema()}},
+ "required": True,
+ },
+ },
+)
+async def create_item(request: Request):
+ raw_body = await request.body()
+ try:
+ data = yaml.safe_load(raw_body)
+ except yaml.YAMLError:
+ raise HTTPException(status_code=422, detail="Invalid YAML")
+ try:
+ item = Item.parse_obj(data)
+ except ValidationError as e:
+ raise HTTPException(status_code=422, detail=e.errors())
+ return item
--- /dev/null
+import yaml
+from fastapi import FastAPI, HTTPException, Request
+from pydantic import BaseModel, ValidationError
+
+app = FastAPI()
+
+
+class Item(BaseModel):
+ name: str
+ tags: list[str]
+
+
+@app.post(
+ "/items/",
+ openapi_extra={
+ "requestBody": {
+ "content": {"application/x-yaml": {"schema": Item.model_json_schema()}},
+ "required": True,
+ },
+ },
+)
+async def create_item(request: Request):
+ raw_body = await request.body()
+ try:
+ data = yaml.safe_load(raw_body)
+ except yaml.YAMLError:
+ raise HTTPException(status_code=422, detail="Invalid YAML")
+ try:
+ item = Item.model_validate(data)
+ except ValidationError as e:
+ raise HTTPException(status_code=422, detail=e.errors(include_url=False))
+ return item
--- /dev/null
+from datetime import datetime
+
+from fastapi import FastAPI
+from fastapi.encoders import jsonable_encoder
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+
+class Item(BaseModel):
+ title: str
+ timestamp: datetime
+ description: str | None = None
+
+
+app = FastAPI()
+
+
+@app.put("/items/{id}")
+def update_item(id: str, item: Item):
+ json_compatible_item_data = jsonable_encoder(item)
+ return JSONResponse(content=json_compatible_item_data)
-from pydantic_settings import BaseSettings
+from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
admin_email: str
items_per_user: int = 50
- class Config:
- env_file = ".env"
+ model_config = SettingsConfigDict(env_file=".env")
--- /dev/null
+from pydantic import BaseSettings
+
+
+class Settings(BaseSettings):
+ app_name: str = "Awesome API"
+ admin_email: str
+ items_per_user: int = 50
+
+ class Config:
+ env_file = ".env"
from functools import lru_cache
-from typing import Annotated
from fastapi import Depends, FastAPI
+from typing_extensions import Annotated
from . import config
-from pydantic_settings import BaseSettings
+from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
admin_email: str
items_per_user: int = 50
- class Config:
- env_file = ".env"
+ model_config = SettingsConfigDict(env_file=".env")
--- /dev/null
+from pydantic import BaseSettings
+
+
+class Settings(BaseSettings):
+ app_name: str = "Awesome API"
+ admin_email: str
+ items_per_user: int = 50
+
+ class Config:
+ env_file = ".env"
from functools import lru_cache
+from typing import Annotated
from fastapi import Depends, FastAPI
-from typing_extensions import Annotated
from . import config
"docs_src/custom_response/tutorial007.py" = ["B007"]
"docs_src/dataclasses/tutorial003.py" = ["I001"]
"docs_src/path_operation_advanced_configuration/tutorial007.py" = ["B904"]
+"docs_src/path_operation_advanced_configuration/tutorial007_py39.py" = ["B904"]
"docs_src/path_operation_advanced_configuration/tutorial007_pv1.py" = ["B904"]
+"docs_src/path_operation_advanced_configuration/tutorial007_pv1_py39.py" = ["B904"]
"docs_src/custom_request_and_route/tutorial002.py" = ["B904"]
+"docs_src/custom_request_and_route/tutorial002_py39.py" = ["B904"]
+"docs_src/custom_request_and_route/tutorial002_py310.py" = ["B904"]
+"docs_src/custom_request_and_route/tutorial002_an.py" = ["B904"]
+"docs_src/custom_request_and_route/tutorial002_an_py39.py" = ["B904"]
+"docs_src/custom_request_and_route/tutorial002_an_py310.py" = ["B904"]
"docs_src/dependencies/tutorial008_an.py" = ["F821"]
"docs_src/dependencies/tutorial008_an_py39.py" = ["F821"]
"docs_src/query_params_str_validations/tutorial012_an.py" = ["B006"]
+import importlib
import os
import shutil
+import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from docs_src.additional_responses.tutorial002 import app
+from tests.utils import needs_py310
-client = TestClient(app)
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial002"),
+ pytest.param("tutorial002_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.additional_responses.{request.param}")
-def test_path_operation():
+ client = TestClient(mod.app)
+ client.headers.clear()
+ return client
+
+
+def test_path_operation(client: TestClient):
response = client.get("/items/foo")
assert response.status_code == 200, response.text
assert response.json() == {"id": "foo", "value": "there goes my hero"}
-def test_path_operation_img():
+def test_path_operation_img(client: TestClient):
shutil.copy("./docs/en/docs/img/favicon.png", "./image.png")
response = client.get("/items/foo?img=1")
assert response.status_code == 200, response.text
os.remove("./image.png")
-def test_openapi_schema():
+def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == {
+import importlib
import os
import shutil
+import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from docs_src.additional_responses.tutorial004 import app
+from tests.utils import needs_py310
-client = TestClient(app)
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial004"),
+ pytest.param("tutorial004_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.additional_responses.{request.param}")
-def test_path_operation():
+ client = TestClient(mod.app)
+ client.headers.clear()
+ return client
+
+
+def test_path_operation(client: TestClient):
response = client.get("/items/foo")
assert response.status_code == 200, response.text
assert response.json() == {"id": "foo", "value": "there goes my hero"}
-def test_path_operation_img():
+def test_path_operation_img(client: TestClient):
shutil.copy("./docs/en/docs/img/favicon.png", "./image.png")
response = client.get("/items/foo?img=1")
assert response.status_code == 200, response.text
os.remove("./image.png")
-def test_openapi_schema():
+def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == {
import gzip
+import importlib
import json
import pytest
from fastapi import Request
from fastapi.testclient import TestClient
-from docs_src.custom_request_and_route.tutorial001 import app
+from tests.utils import needs_py39, needs_py310
-@app.get("/check-class")
-async def check_gzip_request(request: Request):
- return {"request_class": type(request).__name__}
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial001"),
+ pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py310", marks=needs_py310),
+ pytest.param("tutorial001_an"),
+ pytest.param("tutorial001_an_py39", marks=needs_py39),
+ pytest.param("tutorial001_an_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.custom_request_and_route.{request.param}")
+ @mod.app.get("/check-class")
+ async def check_gzip_request(request: Request):
+ return {"request_class": type(request).__name__}
-client = TestClient(app)
+ client = TestClient(mod.app)
+ return client
@pytest.mark.parametrize("compress", [True, False])
-def test_gzip_request(compress):
+def test_gzip_request(client: TestClient, compress):
n = 1000
headers = {}
body = [1] * n
assert response.json() == {"sum": n}
-def test_request_class():
+def test_request_class(client: TestClient):
response = client.get("/check-class")
assert response.json() == {"request_class": "GzipRequest"}
+import importlib
+
+import pytest
from dirty_equals import IsDict, IsOneOf
from fastapi.testclient import TestClient
-from docs_src.custom_request_and_route.tutorial002 import app
+from tests.utils import needs_py39, needs_py310
+
+
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial002"),
+ pytest.param("tutorial002_py39", marks=needs_py39),
+ pytest.param("tutorial002_py310", marks=needs_py310),
+ pytest.param("tutorial002_an"),
+ pytest.param("tutorial002_an_py39", marks=needs_py39),
+ pytest.param("tutorial002_an_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.custom_request_and_route.{request.param}")
-client = TestClient(app)
+ client = TestClient(mod.app)
+ return client
-def test_endpoint_works():
+def test_endpoint_works(client: TestClient):
response = client.post("/", json=[1, 2, 3])
assert response.json() == 6
-def test_exception_handler_body_access():
+def test_exception_handler_body_access(client: TestClient):
response = client.post("/", json={"numbers": [1, 2, 3]})
assert response.json() == IsDict(
{
+import importlib
+
+import pytest
from fastapi.testclient import TestClient
-from docs_src.custom_request_and_route.tutorial003 import app
+from tests.utils import needs_py310
+
+
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial003"),
+ pytest.param("tutorial003_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.custom_request_and_route.{request.param}")
-client = TestClient(app)
+ client = TestClient(mod.app)
+ return client
-def test_get():
+def test_get(client: TestClient):
response = client.get("/")
assert response.json() == {"message": "Not timed"}
assert "X-Response-Time" not in response.headers
-def test_get_timed():
+def test_get_timed(client: TestClient):
response = client.get("/timed")
assert response.json() == {"message": "It's the time of my life"}
assert "X-Response-Time" in response.headers
+import importlib
+
+import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from docs_src.dataclasses.tutorial001 import app
+from tests.utils import needs_py310
+
+
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial001"),
+ pytest.param("tutorial001_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.dataclasses.{request.param}")
-client = TestClient(app)
+ client = TestClient(mod.app)
+ client.headers.clear()
+ return client
-def test_post_item():
+def test_post_item(client: TestClient):
response = client.post("/items/", json={"name": "Foo", "price": 3})
assert response.status_code == 200
assert response.json() == {
}
-def test_post_invalid_item():
+def test_post_invalid_item(client: TestClient):
response = client.post("/items/", json={"name": "Foo", "price": "invalid price"})
assert response.status_code == 422
assert response.json() == IsDict(
)
-def test_openapi_schema():
+def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == {
+import importlib
+
+import pytest
from dirty_equals import IsDict, IsOneOf
from fastapi.testclient import TestClient
-from docs_src.dataclasses.tutorial002 import app
+from tests.utils import needs_py39, needs_py310
+
+
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial002"),
+ pytest.param("tutorial002_py39", marks=needs_py39),
+ pytest.param("tutorial002_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.dataclasses.{request.param}")
-client = TestClient(app)
+ client = TestClient(mod.app)
+ client.headers.clear()
+ return client
-def test_get_item():
+def test_get_item(client: TestClient):
response = client.get("/items/next")
assert response.status_code == 200
assert response.json() == {
}
-def test_openapi_schema():
+def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == {
+import importlib
+
+import pytest
from fastapi.testclient import TestClient
-from docs_src.dataclasses.tutorial003 import app
+from ...utils import needs_py39, needs_py310, needs_pydanticv1, needs_pydanticv2
+
-from ...utils import needs_pydanticv1, needs_pydanticv2
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial003"),
+ pytest.param("tutorial003_py39", marks=needs_py39),
+ pytest.param("tutorial003_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.dataclasses.{request.param}")
-client = TestClient(app)
+ client = TestClient(mod.app)
+ client.headers.clear()
+ return client
-def test_post_authors_item():
+def test_post_authors_item(client: TestClient):
response = client.post(
"/authors/foo/items/",
json=[{"name": "Bar"}, {"name": "Baz", "description": "Drop the Baz"}],
}
-def test_get_authors():
+def test_get_authors(client: TestClient):
response = client.get("/authors/")
assert response.status_code == 200
assert response.json() == [
@needs_pydanticv2
-def test_openapi_schema():
+def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == {
# TODO: remove when deprecating Pydantic v1
@needs_pydanticv1
-def test_openapi_schema_pv1():
+def test_openapi_schema_pv1(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == {
+import importlib
+from types import ModuleType
+
+import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from docs_src.openapi_callbacks.tutorial001 import app, invoice_notification
+from tests.utils import needs_py310
+
+
+@pytest.fixture(
+ name="mod",
+ params=[
+ pytest.param("tutorial001"),
+ pytest.param("tutorial001_py310", marks=needs_py310),
+ ],
+)
+def get_mod(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.openapi_callbacks.{request.param}")
+ return mod
+
-client = TestClient(app)
+@pytest.fixture(name="client")
+def get_client(mod: ModuleType):
+ client = TestClient(mod.app)
+ client.headers.clear()
+ return client
-def test_get():
+def test_get(client: TestClient):
response = client.post(
"/invoices/", json={"id": "fooinvoice", "customer": "John", "total": 5.3}
)
assert response.json() == {"msg": "Invoice received"}
-def test_dummy_callback():
+def test_dummy_callback(mod: ModuleType):
# Just for coverage
- invoice_notification({})
+ mod.invoice_notification({})
-def test_openapi_schema():
+def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == {
+import importlib
+
+import pytest
from fastapi.testclient import TestClient
-from docs_src.path_operation_advanced_configuration.tutorial004 import app
+from ...utils import needs_py39, needs_py310, needs_pydanticv1, needs_pydanticv2
+
-from ...utils import needs_pydanticv1, needs_pydanticv2
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial004"),
+ pytest.param("tutorial004_py39", marks=needs_py39),
+ pytest.param("tutorial004_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(
+ f"docs_src.path_operation_advanced_configuration.{request.param}"
+ )
-client = TestClient(app)
+ client = TestClient(mod.app)
+ client.headers.clear()
+ return client
-def test_query_params_str_validations():
+def test_query_params_str_validations(client: TestClient):
response = client.post("/items/", json={"name": "Foo", "price": 42})
assert response.status_code == 200, response.text
assert response.json() == {
@needs_pydanticv2
-def test_openapi_schema():
+def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == {
# TODO: remove when deprecating Pydantic v1
@needs_pydanticv1
-def test_openapi_schema_pv1():
+def test_openapi_schema_pv1(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == {
+import importlib
+
import pytest
from fastapi.testclient import TestClient
-from ...utils import needs_pydanticv2
+from ...utils import needs_py39, needs_pydanticv2
-@pytest.fixture(name="client")
-def get_client():
- from docs_src.path_operation_advanced_configuration.tutorial007 import app
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial007"),
+ pytest.param("tutorial007_py39", marks=needs_py39),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(
+ f"docs_src.path_operation_advanced_configuration.{request.param}"
+ )
- client = TestClient(app)
+ client = TestClient(mod.app)
return client
+import importlib
+
import pytest
from fastapi.testclient import TestClient
-from ...utils import needs_pydanticv1
+from ...utils import needs_py39, needs_pydanticv1
-@pytest.fixture(name="client")
-def get_client():
- from docs_src.path_operation_advanced_configuration.tutorial007_pv1 import app
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial007_pv1"),
+ pytest.param("tutorial007_pv1_py39", marks=needs_py39),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(
+ f"docs_src.path_operation_advanced_configuration.{request.param}"
+ )
- client = TestClient(app)
+ client = TestClient(mod.app)
return client
--- /dev/null
+import importlib
+
+import pytest
+from fastapi.testclient import TestClient
+
+from ...utils import needs_py310, needs_pydanticv1, needs_pydanticv2
+
+
+@pytest.fixture(
+ name="client",
+ params=[
+ pytest.param("tutorial001"),
+ pytest.param("tutorial001_py310", marks=needs_py310),
+ ],
+)
+def get_client(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.response_directly.{request.param}")
+
+ client = TestClient(mod.app)
+ return client
+
+
+def test_path_operation(client: TestClient):
+ response = client.put(
+ "/items/1",
+ json={
+ "title": "Foo",
+ "timestamp": "2023-01-01T12:00:00",
+ "description": "A test item",
+ },
+ )
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "description": "A test item",
+ "timestamp": "2023-01-01T12:00:00",
+ "title": "Foo",
+ }
+
+
+@needs_pydanticv2
+def test_openapi_schema_pv2(client: TestClient):
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "info": {
+ "title": "FastAPI",
+ "version": "0.1.0",
+ },
+ "openapi": "3.1.0",
+ "paths": {
+ "/items/{id}": {
+ "put": {
+ "operationId": "update_item_items__id__put",
+ "parameters": [
+ {
+ "in": "path",
+ "name": "id",
+ "required": True,
+ "schema": {"title": "Id", "type": "string"},
+ },
+ ],
+ "requestBody": {
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/Item",
+ },
+ },
+ },
+ "required": True,
+ },
+ "responses": {
+ "200": {
+ "content": {
+ "application/json": {"schema": {}},
+ },
+ "description": "Successful Response",
+ },
+ "422": {
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError",
+ },
+ },
+ },
+ "description": "Validation Error",
+ },
+ },
+ "summary": "Update Item",
+ },
+ },
+ },
+ "components": {
+ "schemas": {
+ "HTTPValidationError": {
+ "properties": {
+ "detail": {
+ "items": {
+ "$ref": "#/components/schemas/ValidationError",
+ },
+ "title": "Detail",
+ "type": "array",
+ },
+ },
+ "title": "HTTPValidationError",
+ "type": "object",
+ },
+ "Item": {
+ "properties": {
+ "description": {
+ "anyOf": [
+ {"type": "string"},
+ {"type": "null"},
+ ],
+ "title": "Description",
+ },
+ "timestamp": {
+ "format": "date-time",
+ "title": "Timestamp",
+ "type": "string",
+ },
+ "title": {"title": "Title", "type": "string"},
+ },
+ "required": [
+ "title",
+ "timestamp",
+ ],
+ "title": "Item",
+ "type": "object",
+ },
+ "ValidationError": {
+ "properties": {
+ "loc": {
+ "items": {
+ "anyOf": [
+ {"type": "string"},
+ {"type": "integer"},
+ ],
+ },
+ "title": "Location",
+ "type": "array",
+ },
+ "msg": {"title": "Message", "type": "string"},
+ "type": {"title": "Error Type", "type": "string"},
+ },
+ "required": ["loc", "msg", "type"],
+ "title": "ValidationError",
+ "type": "object",
+ },
+ },
+ },
+ }
+
+
+@needs_pydanticv1
+def test_openapi_schema_pv1(client: TestClient):
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "info": {
+ "title": "FastAPI",
+ "version": "0.1.0",
+ },
+ "openapi": "3.1.0",
+ "paths": {
+ "/items/{id}": {
+ "put": {
+ "operationId": "update_item_items__id__put",
+ "parameters": [
+ {
+ "in": "path",
+ "name": "id",
+ "required": True,
+ "schema": {
+ "title": "Id",
+ "type": "string",
+ },
+ },
+ ],
+ "requestBody": {
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/Item",
+ },
+ },
+ },
+ "required": True,
+ },
+ "responses": {
+ "200": {
+ "content": {
+ "application/json": {
+ "schema": {},
+ },
+ },
+ "description": "Successful Response",
+ },
+ "422": {
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError",
+ },
+ },
+ },
+ "description": "Validation Error",
+ },
+ },
+ "summary": "Update Item",
+ },
+ },
+ },
+ "components": {
+ "schemas": {
+ "HTTPValidationError": {
+ "properties": {
+ "detail": {
+ "items": {
+ "$ref": "#/components/schemas/ValidationError",
+ },
+ "title": "Detail",
+ "type": "array",
+ },
+ },
+ "title": "HTTPValidationError",
+ "type": "object",
+ },
+ "Item": {
+ "properties": {
+ "description": {
+ "title": "Description",
+ "type": "string",
+ },
+ "timestamp": {
+ "format": "date-time",
+ "title": "Timestamp",
+ "type": "string",
+ },
+ "title": {
+ "title": "Title",
+ "type": "string",
+ },
+ },
+ "required": [
+ "title",
+ "timestamp",
+ ],
+ "title": "Item",
+ "type": "object",
+ },
+ "ValidationError": {
+ "properties": {
+ "loc": {
+ "items": {
+ "anyOf": [
+ {
+ "type": "string",
+ },
+ {
+ "type": "integer",
+ },
+ ],
+ },
+ "title": "Location",
+ "type": "array",
+ },
+ "msg": {
+ "title": "Message",
+ "type": "string",
+ },
+ "type": {
+ "title": "Error Type",
+ "type": "string",
+ },
+ },
+ "required": [
+ "loc",
+ "msg",
+ "type",
+ ],
+ "title": "ValidationError",
+ "type": "object",
+ },
+ },
+ },
+ }
+import importlib
+from types import ModuleType
+
+import pytest
from pytest import MonkeyPatch
-from ...utils import needs_pydanticv2
+from ...utils import needs_py39, needs_pydanticv2
-@needs_pydanticv2
-def test_settings(monkeypatch: MonkeyPatch):
- from docs_src.settings.app02 import main
+@pytest.fixture(
+ name="mod_path",
+ params=[
+ pytest.param("app02"),
+ pytest.param("app02_an"),
+ pytest.param("app02_an_py39", marks=needs_py39),
+ ],
+)
+def get_mod_path(request: pytest.FixtureRequest):
+ mod_path = f"docs_src.settings.{request.param}"
+ return mod_path
+
+
+@pytest.fixture(name="main_mod")
+def get_main_mod(mod_path: str) -> ModuleType:
+ main_mod = importlib.import_module(f"{mod_path}.main")
+ return main_mod
+
+@pytest.fixture(name="test_main_mod")
+def get_test_main_mod(mod_path: str) -> ModuleType:
+ test_main_mod = importlib.import_module(f"{mod_path}.test_main")
+ return test_main_mod
+
+
+@needs_pydanticv2
+def test_settings(main_mod: ModuleType, monkeypatch: MonkeyPatch):
monkeypatch.setenv("ADMIN_EMAIL", "admin@example.com")
- settings = main.get_settings()
+ settings = main_mod.get_settings()
assert settings.app_name == "Awesome API"
assert settings.items_per_user == 50
@needs_pydanticv2
-def test_override_settings():
- from docs_src.settings.app02 import test_main
-
- test_main.test_app()
+def test_override_settings(test_main_mod: ModuleType):
+ test_main_mod.test_app()
--- /dev/null
+import importlib
+from types import ModuleType
+
+import pytest
+from fastapi.testclient import TestClient
+from pytest import MonkeyPatch
+
+from ...utils import needs_py39, needs_pydanticv1, needs_pydanticv2
+
+
+@pytest.fixture(
+ name="mod_path",
+ params=[
+ pytest.param("app03"),
+ pytest.param("app03_an"),
+ pytest.param("app03_an_py39", marks=needs_py39),
+ ],
+)
+def get_mod_path(request: pytest.FixtureRequest):
+ mod_path = f"docs_src.settings.{request.param}"
+ return mod_path
+
+
+@pytest.fixture(name="main_mod")
+def get_main_mod(mod_path: str) -> ModuleType:
+ main_mod = importlib.import_module(f"{mod_path}.main")
+ return main_mod
+
+
+@needs_pydanticv2
+def test_settings(main_mod: ModuleType, monkeypatch: MonkeyPatch):
+ monkeypatch.setenv("ADMIN_EMAIL", "admin@example.com")
+ settings = main_mod.get_settings()
+ assert settings.app_name == "Awesome API"
+ assert settings.admin_email == "admin@example.com"
+ assert settings.items_per_user == 50
+
+
+@needs_pydanticv1
+def test_settings_pv1(mod_path: str, monkeypatch: MonkeyPatch):
+ monkeypatch.setenv("ADMIN_EMAIL", "admin@example.com")
+ config_mod = importlib.import_module(f"{mod_path}.config_pv1")
+ settings = config_mod.Settings()
+ assert settings.app_name == "Awesome API"
+ assert settings.admin_email == "admin@example.com"
+ assert settings.items_per_user == 50
+
+
+@needs_pydanticv2
+def test_endpoint(main_mod: ModuleType, monkeypatch: MonkeyPatch):
+ monkeypatch.setenv("ADMIN_EMAIL", "admin@example.com")
+ client = TestClient(main_mod.app)
+ response = client.get("/info")
+ assert response.status_code == 200
+ assert response.json() == {
+ "app_name": "Awesome API",
+ "admin_email": "admin@example.com",
+ "items_per_user": 50,
+ }
--- /dev/null
+from fastapi.testclient import TestClient
+
+from docs_src.using_request_directly.tutorial001 import app
+
+client = TestClient(app)
+
+
+def test_path_operation():
+ response = client.get("/items/foo")
+ assert response.status_code == 200
+ assert response.json() == {"client_host": "testclient", "item_id": "foo"}
+
+
+def test_openapi():
+ response = client.get("/openapi.json")
+ assert response.status_code == 200
+ assert response.json() == {
+ "info": {
+ "title": "FastAPI",
+ "version": "0.1.0",
+ },
+ "openapi": "3.1.0",
+ "paths": {
+ "/items/{item_id}": {
+ "get": {
+ "operationId": "read_root_items__item_id__get",
+ "parameters": [
+ {
+ "in": "path",
+ "name": "item_id",
+ "required": True,
+ "schema": {
+ "title": "Item Id",
+ "type": "string",
+ },
+ },
+ ],
+ "responses": {
+ "200": {
+ "content": {
+ "application/json": {
+ "schema": {},
+ },
+ },
+ "description": "Successful Response",
+ },
+ "422": {
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError",
+ },
+ },
+ },
+ "description": "Validation Error",
+ },
+ },
+ "summary": "Read Root",
+ },
+ },
+ },
+ "components": {
+ "schemas": {
+ "HTTPValidationError": {
+ "properties": {
+ "detail": {
+ "items": {
+ "$ref": "#/components/schemas/ValidationError",
+ },
+ "title": "Detail",
+ "type": "array",
+ },
+ },
+ "title": "HTTPValidationError",
+ "type": "object",
+ },
+ "ValidationError": {
+ "properties": {
+ "loc": {
+ "items": {
+ "anyOf": [
+ {
+ "type": "string",
+ },
+ {
+ "type": "integer",
+ },
+ ],
+ },
+ "title": "Location",
+ "type": "array",
+ },
+ "msg": {
+ "title": "Message",
+ "type": "string",
+ },
+ "type": {
+ "title": "Error Type",
+ "type": "string",
+ },
+ },
+ "required": [
+ "loc",
+ "msg",
+ "type",
+ ],
+ "title": "ValidationError",
+ "type": "object",
+ },
+ },
+ },
+ }
+import importlib
+from types import ModuleType
+
+import pytest
from fastapi.testclient import TestClient
-from docs_src.websockets.tutorial003 import app, html
+from ...utils import needs_py39
+
+
+@pytest.fixture(
+ name="mod",
+ params=[
+ pytest.param("tutorial003"),
+ pytest.param("tutorial003_py39", marks=needs_py39),
+ ],
+)
+def get_mod(request: pytest.FixtureRequest):
+ mod = importlib.import_module(f"docs_src.websockets.{request.param}")
+
+ return mod
+
+
+@pytest.fixture(name="html")
+def get_html(mod: ModuleType):
+ return mod.html
+
+
+@pytest.fixture(name="client")
+def get_client(mod: ModuleType):
+ client = TestClient(mod.app)
-client = TestClient(app)
+ return client
-def test_get():
+@needs_py39
+def test_get(client: TestClient, html: str):
response = client.get("/")
assert response.text == html
-def test_websocket_handle_disconnection():
+@needs_py39
+def test_websocket_handle_disconnection(client: TestClient):
with client.websocket_connect("/ws/1234") as connection, client.websocket_connect(
"/ws/5678"
) as connection_two:
+++ /dev/null
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-from ...utils import needs_py39
-
-
-@pytest.fixture(name="app")
-def get_app():
- from docs_src.websockets.tutorial003_py39 import app
-
- return app
-
-
-@pytest.fixture(name="html")
-def get_html():
- from docs_src.websockets.tutorial003_py39 import html
-
- return html
-
-
-@pytest.fixture(name="client")
-def get_client(app: FastAPI):
- client = TestClient(app)
-
- return client
-
-
-@needs_py39
-def test_get(client: TestClient, html: str):
- response = client.get("/")
- assert response.text == html
-
-
-@needs_py39
-def test_websocket_handle_disconnection(client: TestClient):
- with client.websocket_connect("/ws/1234") as connection, client.websocket_connect(
- "/ws/5678"
- ) as connection_two:
- connection.send_text("Hello from 1234")
- data1 = connection.receive_text()
- assert data1 == "You wrote: Hello from 1234"
- data2 = connection_two.receive_text()
- client1_says = "Client #1234 says: Hello from 1234"
- assert data2 == client1_says
- data1 = connection.receive_text()
- assert data1 == client1_says
- connection_two.close()
- data1 = connection.receive_text()
- assert data1 == "Client #5678 left the chat"
projects / thirdparty / fastapi / fastapi.git / commitdiff
