Compare commits

...

25 Commits

Author SHA1 Message Date
Sebastián Ramírez
7b0b915749 🔖 Release version 0.124.2 2025-12-10 13:07:53 +01:00
github-actions[bot]
96bdde376f 📝 Update release notes
[skip ci]
2025-12-10 12:06:32 +00:00
Sebastián Ramírez
7ba042e069 🐛 Fix support for if TYPE_CHECKING, non-evaluated stringified annotations (#14485) 2025-12-10 13:06:05 +01:00
Sebastián Ramírez
60699f306b 🔖 Release version 0.124.1 2025-12-10 11:38:41 +01:00
github-actions[bot]
ae7af59c6d 📝 Update release notes
[skip ci]
2025-12-10 10:36:56 +00:00
Sebastián Ramírez
42b250d14d 🐛 Fix handling arbitrary types when using arbitrary_types_allowed=True (#14482) 2025-12-10 11:36:29 +01:00
github-actions[bot]
71a17b5932 📝 Update release notes
[skip ci]
2025-12-10 08:55:57 +00:00
Motov Yurii
9475024640 📝 Add variants for code examples in "Advanced User Guide" (#14413) 2025-12-10 09:55:32 +01:00
github-actions[bot]
5b28a04d55 📝 Update release notes
[skip ci]
2025-12-09 11:12:49 +00:00
Sebastián Ramírez
8cedb742cb Add test for Pydantic v2, dataclasses, UUID, and __annotations__ (#14477) 2025-12-09 12:12:24 +01:00
github-actions[bot]
320e7ce8fd 📝 Update release notes
[skip ci]
2025-12-08 13:05:20 +00:00
Alejandra
81517f66cc 📝 Update tech stack in project generation docs (#14472) 2025-12-08 13:04:54 +00:00
Sebastián Ramírez
b5ca13249e 🔖 Release version 0.124.0 2025-12-06 14:09:51 +01:00
github-actions[bot]
a2cef707e3 📝 Update release notes
[skip ci]
2025-12-06 12:23:23 +00:00
Yuji Teshima
5b6245666b ✏️ Fix typo in scripts/mkdocs_hooks.py (#14457) 2025-12-06 13:23:01 +01:00
github-actions[bot]
dbd34f1578 📝 Update release notes
[skip ci]
2025-12-06 12:22:24 +00:00
Savannah Ostrowski
e1117f7550 🚸 Improve tracebacks by adding endpoint metadata (#14306)
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
2025-12-06 12:21:57 +00:00
Sebastián Ramírez
08b09e5236 🔖 Release version 0.123.10 2025-12-05 22:26:36 +01:00
github-actions[bot]
e7d7038dfa 📝 Update release notes
[skip ci]
2025-12-05 21:21:29 +00:00
Motov Yurii
da0ffab0b2 🐛 Fix using class (not instance) dependency that has __call__ method (#14458)
Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
2025-12-05 21:21:05 +00:00
github-actions[bot]
516169428d 📝 Update release notes
[skip ci]
2025-12-05 20:19:54 +00:00
Motov Yurii
812a1926f0 🐛 Fix separate_input_output_schemas=False with computed_field (#14453) 2025-12-05 21:19:30 +01:00
Sebastián Ramírez
f0dd1046a6 🔖 Release version 0.123.9 2025-12-04 23:23:21 +01:00
github-actions[bot]
188d631011 📝 Update release notes
[skip ci]
2025-12-04 22:22:25 +00:00
Sebastián Ramírez
0b5fa563cd 🐛 Fix OAuth2 scopes in OpenAPI in extra corner cases, parent dependency with scopes, sub-dependency security scheme without scopes (#14459) 2025-12-04 23:22:01 +01:00
82 changed files with 2716 additions and 372 deletions

View File

@@ -14,7 +14,7 @@ GitHub-Repository: <a href="https://github.com/tiangolo/full-stack-fastapi-templ
- 💾 [PostgreSQL](https://www.postgresql.org) als SQL-Datenbank.
- 🚀 [React](https://react.dev) für das Frontend.
- 💃 Verwendung von TypeScript, Hooks, [Vite](https://vitejs.dev) und anderen Teilen eines modernen Frontend-Stacks.
- 🎨 [Chakra UI](https://chakra-ui.com) für die Frontend-Komponenten.
- 🎨 [Tailwind CSS](https://tailwindcss.com) und [shadcn/ui](https://ui.shadcn.com) für die Frontend-Komponenten.
- 🤖 Ein automatisch generierter Frontend-Client.
- 🧪 [Playwright](https://playwright.dev) für End-to-End-Tests.
- 🦇 Unterstützung des Dunkelmodus.

View File

@@ -175,7 +175,7 @@ You can use this same `responses` parameter to add different media types for the
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
@@ -237,7 +237,7 @@ You can use that technique to reuse some predefined responses in your *path oper
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 }

View File

@@ -4,7 +4,7 @@ FastAPI is built on top of **Pydantic**, and I have been showing you how to use
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>.
@@ -32,7 +32,7 @@ But if you have a bunch of dataclasses laying around, this is a nice trick to us
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.
@@ -48,7 +48,7 @@ In some cases, you might still have to use Pydantic's version of `dataclasses`.
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`.

View File

@@ -31,7 +31,7 @@ It will have a *path operation* that will receive an `Invoice` body, and a query
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
@@ -90,7 +90,7 @@ Temporarily adopting this point of view (of the *external developer*) can help y
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 }
@@ -101,7 +101,7 @@ It should look just like a normal FastAPI *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*:
@@ -169,7 +169,7 @@ At this point you have the *callback path operation(s)* needed (the one(s) that
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

View File

@@ -50,7 +50,7 @@ Adding an `\f` (an escaped "form feed" character) causes **FastAPI** to truncate
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 }
@@ -155,13 +155,13 @@ For example, in this application we don't use FastAPI's integrated functionality
//// 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] *}
////
@@ -179,13 +179,13 @@ And then in our code, we parse that YAML content directly, and then we are again
//// 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] *}
////

View File

@@ -34,7 +34,7 @@ For example, you cannot put a Pydantic model in a `JSONResponse` without first c
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

View File

@@ -148,7 +148,7 @@ This could be especially useful during testing, as it's very easy to override a
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()`.
@@ -174,7 +174,7 @@ And then we can require it from the *path operation function* as a dependency an
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.
@@ -217,7 +217,7 @@ And then update your `config.py` with:
//// tab | Pydantic v2
{* ../../docs_src/settings/app03_an/config.py hl[9] *}
{* ../../docs_src/settings/app03_an_py39/config.py hl[9] *}
/// tip
@@ -229,7 +229,7 @@ The `model_config` attribute is used just for Pydantic configuration. You can re
//// 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

View File

@@ -40,7 +40,7 @@ FastAPI includes some default configuration parameters appropriate for most of t
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`.

View File

@@ -42,7 +42,7 @@ If there's no `gzip` in the header, it will not try to decompress the body.
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 }
@@ -54,7 +54,7 @@ This method returns a function. And that function is what will receive a request
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
@@ -92,18 +92,18 @@ We can also use this same approach to access the request body in an exception ha
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] *}

View File

@@ -9,18 +9,18 @@ GitHub Repository: <a href="https://github.com/tiangolo/full-stack-fastapi-templ
## Full Stack FastAPI Template - Technology Stack and Features { #full-stack-fastapi-template-technology-stack-and-features }
- ⚡ [**FastAPI**](https://fastapi.tiangolo.com) for the Python backend API.
- 🧰 [SQLModel](https://sqlmodel.tiangolo.com) for the Python SQL database interactions (ORM).
- 🔍 [Pydantic](https://docs.pydantic.dev), used by FastAPI, for the data validation and settings management.
- 💾 [PostgreSQL](https://www.postgresql.org) as the SQL database.
- 🧰 [SQLModel](https://sqlmodel.tiangolo.com) for the Python SQL database interactions (ORM).
- 🔍 [Pydantic](https://docs.pydantic.dev), used by FastAPI, for the data validation and settings management.
- 💾 [PostgreSQL](https://www.postgresql.org) as the SQL database.
- 🚀 [React](https://react.dev) for the frontend.
- 💃 Using TypeScript, hooks, [Vite](https://vitejs.dev), and other parts of a modern frontend stack.
- 🎨 [Chakra UI](https://chakra-ui.com) for the frontend components.
- 🤖 An automatically generated frontend client.
- 🧪 [Playwright](https://playwright.dev) for End-to-End testing.
- 🦇 Dark mode support.
- 💃 Using TypeScript, hooks, Vite, and other parts of a modern frontend stack.
- 🎨 [Tailwind CSS](https://tailwindcss.com) and [shadcn/ui](https://ui.shadcn.com) for the frontend components.
- 🤖 An automatically generated frontend client.
- 🧪 [Playwright](https://playwright.dev) for End-to-End testing.
- 🦇 Dark mode support.
- 🐋 [Docker Compose](https://www.docker.com) for development and production.
- 🔒 Secure password hashing by default.
- 🔑 JWT token authentication.
- 🔑 JWT (JSON Web Token) authentication.
- 📫 Email based password recovery.
- ✅ Tests with [Pytest](https://pytest.org).
- 📞 [Traefik](https://traefik.io) as a reverse proxy / load balancer.

View File

@@ -7,6 +7,50 @@ hide:
## Latest Changes
## 0.124.2
### Fixes
* 🐛 Fix support for `if TYPE_CHECKING`, non-evaluated stringified annotations. PR [#14485](https://github.com/fastapi/fastapi/pull/14485) by [@tiangolo](https://github.com/tiangolo).
## 0.124.1
### Fixes
* 🐛 Fix handling arbitrary types when using `arbitrary_types_allowed=True`. PR [#14482](https://github.com/fastapi/fastapi/pull/14482) by [@tiangolo](https://github.com/tiangolo).
### Docs
* 📝 Add variants for code examples in "Advanced User Guide". PR [#14413](https://github.com/fastapi/fastapi/pull/14413) by [@YuriiMotov](https://github.com/YuriiMotov).
* 📝 Update tech stack in project generation docs. PR [#14472](https://github.com/fastapi/fastapi/pull/14472) by [@alejsdev](https://github.com/alejsdev).
### Internal
* ✅ Add test for Pydantic v2, dataclasses, UUID, and `__annotations__`. PR [#14477](https://github.com/fastapi/fastapi/pull/14477) by [@tiangolo](https://github.com/tiangolo).
## 0.124.0
### Features
* 🚸 Improve tracebacks by adding endpoint metadata. PR [#14306](https://github.com/fastapi/fastapi/pull/14306) by [@savannahostrowski](https://github.com/savannahostrowski).
### Internal
* ✏️ Fix typo in `scripts/mkdocs_hooks.py`. PR [#14457](https://github.com/fastapi/fastapi/pull/14457) by [@yujiteshima](https://github.com/yujiteshima).
## 0.123.10
### Fixes
* 🐛 Fix using class (not instance) dependency that has `__call__` method. PR [#14458](https://github.com/fastapi/fastapi/pull/14458) by [@YuriiMotov](https://github.com/YuriiMotov).
* 🐛 Fix `separate_input_output_schemas=False` with `computed_field`. PR [#14453](https://github.com/fastapi/fastapi/pull/14453) by [@YuriiMotov](https://github.com/YuriiMotov).
## 0.123.9
### Fixes
* 🐛 Fix OAuth2 scopes in OpenAPI in extra corner cases, parent dependency with scopes, sub-dependency security scheme without scopes. PR [#14459](https://github.com/fastapi/fastapi/pull/14459) by [@tiangolo](https://github.com/tiangolo).
## 0.123.8
### Fixes

View File

@@ -85,9 +85,7 @@ You can create the *path operations* for that module using `APIRouter`.
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 }
@@ -95,9 +93,7 @@ And then you use it to declare your *path operations*.
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.
@@ -121,35 +117,7 @@ So we put them in their own `dependencies` module (`app/dependencies.py`).
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
@@ -181,9 +149,7 @@ We know all the *path operations* in this module have the same:
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:
@@ -242,9 +208,7 @@ And we need to get the dependency function from the module `app.dependencies`, t
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 }
@@ -315,9 +279,7 @@ We are not adding the prefix `/items` nor the `tags=["items"]` to each *path ope
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
@@ -343,17 +305,13 @@ You import and create a `FastAPI` class as normally.
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".
@@ -416,17 +374,13 @@ the `router` from `users` would overwrite the one from `items` and we wouldn't b
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
@@ -466,17 +420,13 @@ It contains an `APIRouter` with some admin *path operations* that your organizat
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.
@@ -497,9 +447,7 @@ We can also add *path operations* directly to the `FastAPI` app.
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()`.

View File

@@ -50,7 +50,7 @@ Your API now has the power to control its own <abbr title="This is a joke, just
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.

View File

@@ -121,63 +121,13 @@ It has a `POST` operation that could return several errors.
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.

View File

@@ -14,7 +14,7 @@ Repositório GitHub: <a href="https://github.com/tiangolo/full-stack-fastapi-tem
- 💾 [PostgreSQL](https://www.postgresql.org) como banco de dados SQL.
- 🚀 [React](https://react.dev) para o frontend.
- 💃 Usando TypeScript, hooks, [Vite](https://vitejs.dev), e outras partes de uma _stack_ frontend moderna.
- 🎨 [Chakra UI](https://chakra-ui.com) para os componentes de frontend.
- 🎨 [Tailwind CSS](https://tailwindcss.com) e [shadcn/ui](https://ui.shadcn.com) para os componentes de frontend.
- 🤖 Um cliente frontend automaticamente gerado.
- 🧪 [Playwright](https://playwright.dev) para testes Ponta-a-Ponta.
- 🦇 Suporte para modo escuro.

View File

@@ -0,0 +1,28 @@
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"}

View File

@@ -0,0 +1,30 @@
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"}

View File

@@ -0,0 +1,36 @@
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)}

View File

@@ -0,0 +1,36 @@
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)}

View File

@@ -0,0 +1,35 @@
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)}

View File

@@ -0,0 +1,35 @@
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)}

View File

@@ -0,0 +1,35 @@
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)}

View File

@@ -0,0 +1,30 @@
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)

View File

@@ -0,0 +1,30 @@
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)

View File

@@ -0,0 +1,29 @@
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)

View File

@@ -0,0 +1,29 @@
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)

View File

@@ -0,0 +1,29 @@
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)

View File

@@ -0,0 +1,39 @@
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)

View File

@@ -0,0 +1,19 @@
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

View File

@@ -0,0 +1,25 @@
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"],
}

View File

@@ -0,0 +1,26 @@
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"],
}

View File

@@ -0,0 +1,54 @@
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",
},
],
},
]

View File

@@ -0,0 +1,55 @@
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",
},
],
},
]

View File

@@ -0,0 +1,51 @@
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"}

View File

@@ -0,0 +1,28 @@
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

View File

@@ -0,0 +1,30 @@
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

View File

@@ -0,0 +1,32 @@
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

View File

@@ -0,0 +1,32 @@
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

View File

@@ -0,0 +1,21 @@
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)

View File

@@ -1,4 +1,4 @@
from pydantic_settings import BaseSettings
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
@@ -6,5 +6,4 @@ class Settings(BaseSettings):
admin_email: str
items_per_user: int = 50
class Config:
env_file = ".env"
model_config = SettingsConfigDict(env_file=".env")

View File

@@ -0,0 +1,10 @@
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"

View File

@@ -1,7 +1,7 @@
from functools import lru_cache
from typing import Annotated
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
from . import config

View File

@@ -1,4 +1,4 @@
from pydantic_settings import BaseSettings
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
@@ -6,5 +6,4 @@ class Settings(BaseSettings):
admin_email: str
items_per_user: int = 50
class Config:
env_file = ".env"
model_config = SettingsConfigDict(env_file=".env")

View File

@@ -0,0 +1,10 @@
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"

View File

@@ -1,7 +1,7 @@
from functools import lru_cache
from typing import Annotated
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
from . import config

View File

@@ -1,6 +1,6 @@
"""FastAPI framework, high performance, easy to learn, fast to code, ready for production"""
__version__ = "0.123.8"
__version__ = "0.124.2"
from starlette import status as status

View File

@@ -1,7 +1,7 @@
import re
import warnings
from copy import copy, deepcopy
from dataclasses import dataclass
from dataclasses import dataclass, is_dataclass
from enum import Enum
from typing import (
Any,
@@ -18,7 +18,7 @@ from typing import (
from fastapi._compat import may_v1, shared
from fastapi.openapi.constants import REF_TEMPLATE
from fastapi.types import IncEx, ModelNameMap, UnionType
from pydantic import BaseModel, TypeAdapter, create_model
from pydantic import BaseModel, ConfigDict, TypeAdapter, create_model
from pydantic import PydanticSchemaGenerationError as PydanticSchemaGenerationError
from pydantic import PydanticUndefinedAnnotation as PydanticUndefinedAnnotation
from pydantic import ValidationError as ValidationError
@@ -64,6 +64,7 @@ class ModelField:
field_info: FieldInfo
name: str
mode: Literal["validation", "serialization"] = "validation"
config: Union[ConfigDict, None] = None
@property
def alias(self) -> str:
@@ -94,8 +95,14 @@ class ModelField:
warnings.simplefilter(
"ignore", category=UnsupportedFieldAttributeWarning
)
annotated_args = (
self.field_info.annotation,
*self.field_info.metadata,
self.field_info,
)
self._type_adapter: TypeAdapter[Any] = TypeAdapter(
Annotated[self.field_info.annotation, self.field_info]
Annotated[annotated_args],
config=self.config,
)
def get_default(self) -> Any:
@@ -171,6 +178,13 @@ def _get_model_config(model: BaseModel) -> Any:
return model.model_config
def _has_computed_fields(field: ModelField) -> bool:
computed_fields = field._type_adapter.core_schema.get("schema", {}).get(
"computed_fields", []
)
return len(computed_fields) > 0
def get_schema_from_model_field(
*,
field: ModelField,
@@ -180,12 +194,9 @@ def get_schema_from_model_field(
],
separate_input_output_schemas: bool = True,
) -> Dict[str, Any]:
computed_fields = field._type_adapter.core_schema.get("schema", {}).get(
"computed_fields", []
)
override_mode: Union[Literal["validation"], None] = (
None
if (separate_input_output_schemas or len(computed_fields) > 0)
if (separate_input_output_schemas or _has_computed_fields(field))
else "validation"
)
# This expects that GenerateJsonSchema was already used to generate the definitions
@@ -208,15 +219,7 @@ def get_definitions(
Dict[Tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue],
Dict[str, Dict[str, Any]],
]:
has_computed_fields: bool = any(
field._type_adapter.core_schema.get("schema", {}).get("computed_fields", [])
for field in fields
)
schema_generator = GenerateJsonSchema(ref_template=REF_TEMPLATE)
override_mode: Union[Literal["validation"], None] = (
None if (separate_input_output_schemas or has_computed_fields) else "validation"
)
validation_fields = [field for field in fields if field.mode == "validation"]
serialization_fields = [field for field in fields if field.mode == "serialization"]
flat_validation_models = get_flat_models_from_fields(
@@ -246,9 +249,16 @@ def get_definitions(
unique_flat_model_fields = {
f for f in flat_model_fields if f.type_ not in input_types
}
inputs = [
(field, override_mode or field.mode, field._type_adapter.core_schema)
(
field,
(
field.mode
if (separate_input_output_schemas or _has_computed_fields(field))
else "validation"
),
field._type_adapter.core_schema,
)
for field in list(fields) + list(unique_flat_model_fields)
]
field_mapping, definitions = schema_generator.generate_definitions(inputs=inputs)
@@ -409,10 +419,21 @@ def create_body_model(
def get_model_fields(model: Type[BaseModel]) -> List[ModelField]:
return [
ModelField(field_info=field_info, name=name)
for name, field_info in model.model_fields.items()
]
model_fields: List[ModelField] = []
for name, field_info in model.model_fields.items():
type_ = field_info.annotation
if lenient_issubclass(type_, (BaseModel, dict)) or is_dataclass(type_):
model_config = None
else:
model_config = model.model_config
model_fields.append(
ModelField(
field_info=field_info,
name=name,
config=model_config,
)
)
return model_fields
# Duplicate of several schema functions from Pydantic v1 to make them compatible with

View File

@@ -2,7 +2,7 @@ import inspect
import sys
from dataclasses import dataclass, field
from functools import cached_property, partial
from typing import Any, Callable, List, Optional, Sequence, Union
from typing import Any, Callable, List, Optional, Union
from fastapi._compat import ModelField
from fastapi.security.base import SecurityBase
@@ -28,12 +28,6 @@ def _impartial(func: Callable[..., Any]) -> Callable[..., Any]:
return func
@dataclass
class SecurityRequirement:
security_scheme: SecurityBase
scopes: Optional[Sequence[str]] = None
@dataclass
class Dependant:
path_params: List[ModelField] = field(default_factory=list)
@@ -42,7 +36,6 @@ class Dependant:
cookie_params: List[ModelField] = field(default_factory=list)
body_params: List[ModelField] = field(default_factory=list)
dependencies: List["Dependant"] = field(default_factory=list)
security_requirements: List[SecurityRequirement] = field(default_factory=list)
name: Optional[str] = None
call: Optional[Callable[..., Any]] = None
request_param_name: Optional[str] = None
@@ -83,11 +76,32 @@ class Dependant:
return True
if self.security_scopes_param_name is not None:
return True
if self._is_security_scheme:
return True
for sub_dep in self.dependencies:
if sub_dep._uses_scopes:
return True
return False
@cached_property
def _is_security_scheme(self) -> bool:
if self.call is None:
return False # pragma: no cover
unwrapped = _unwrapped_call(self.call)
return isinstance(unwrapped, SecurityBase)
# Mainly to get the type of SecurityBase, but it's the same self.call
@cached_property
def _security_scheme(self) -> SecurityBase:
unwrapped = _unwrapped_call(self.call)
assert isinstance(unwrapped, SecurityBase)
return unwrapped
@cached_property
def _security_dependencies(self) -> List["Dependant"]:
security_deps = [dep for dep in self.dependencies if dep._is_security_scheme]
return security_deps
@cached_property
def is_gen_callable(self) -> bool:
if self.call is None:
@@ -96,6 +110,8 @@ class Dependant:
_impartial(self.call)
) or inspect.isgeneratorfunction(_unwrapped_call(self.call)):
return True
if inspect.isclass(_unwrapped_call(self.call)):
return False
dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
if dunder_call is None:
return False # pragma: no cover
@@ -120,6 +136,8 @@ class Dependant:
_impartial(self.call)
) or inspect.isasyncgenfunction(_unwrapped_call(self.call)):
return True
if inspect.isclass(_unwrapped_call(self.call)):
return False
dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
if dunder_call is None:
return False # pragma: no cover
@@ -148,6 +166,8 @@ class Dependant:
_unwrapped_call(self.call)
):
return True
if inspect.isclass(_unwrapped_call(self.call)):
return False
dunder_call = getattr(_impartial(self.call), "__call__", None) # noqa: B004
if dunder_call is None:
return False # pragma: no cover
@@ -162,7 +182,6 @@ class Dependant:
_impartial(dunder_unwrapped_call)
) or iscoroutinefunction(_unwrapped_call(dunder_unwrapped_call)):
return True
# if inspect.isclass(self.call): False, covered by default return
return False
@cached_property

View File

@@ -55,10 +55,9 @@ from fastapi.concurrency import (
asynccontextmanager,
contextmanager_in_threadpool,
)
from fastapi.dependencies.models import Dependant, SecurityRequirement
from fastapi.dependencies.models import Dependant
from fastapi.exceptions import DependencyScopeError
from fastapi.logger import logger
from fastapi.security.base import SecurityBase
from fastapi.security.oauth2 import SecurityScopes
from fastapi.types import DependencyCacheKey
from fastapi.utils import create_model_field, get_path_param_names
@@ -142,10 +141,14 @@ def get_flat_dependant(
*,
skip_repeats: bool = False,
visited: Optional[List[DependencyCacheKey]] = None,
parent_oauth_scopes: Optional[List[str]] = None,
) -> Dependant:
if visited is None:
visited = []
visited.append(dependant.cache_key)
use_parent_oauth_scopes = (parent_oauth_scopes or []) + (
dependant.oauth_scopes or []
)
flat_dependant = Dependant(
path_params=dependant.path_params.copy(),
@@ -153,22 +156,37 @@ def get_flat_dependant(
header_params=dependant.header_params.copy(),
cookie_params=dependant.cookie_params.copy(),
body_params=dependant.body_params.copy(),
security_requirements=dependant.security_requirements.copy(),
name=dependant.name,
call=dependant.call,
request_param_name=dependant.request_param_name,
websocket_param_name=dependant.websocket_param_name,
http_connection_param_name=dependant.http_connection_param_name,
response_param_name=dependant.response_param_name,
background_tasks_param_name=dependant.background_tasks_param_name,
security_scopes_param_name=dependant.security_scopes_param_name,
own_oauth_scopes=dependant.own_oauth_scopes,
parent_oauth_scopes=use_parent_oauth_scopes,
use_cache=dependant.use_cache,
path=dependant.path,
scope=dependant.scope,
)
for sub_dependant in dependant.dependencies:
if skip_repeats and sub_dependant.cache_key in visited:
continue
flat_sub = get_flat_dependant(
sub_dependant, skip_repeats=skip_repeats, visited=visited
sub_dependant,
skip_repeats=skip_repeats,
visited=visited,
parent_oauth_scopes=flat_dependant.oauth_scopes,
)
flat_dependant.dependencies.append(flat_sub)
flat_dependant.path_params.extend(flat_sub.path_params)
flat_dependant.query_params.extend(flat_sub.query_params)
flat_dependant.header_params.extend(flat_sub.header_params)
flat_dependant.cookie_params.extend(flat_sub.cookie_params)
flat_dependant.body_params.extend(flat_sub.body_params)
flat_dependant.security_requirements.extend(flat_sub.security_requirements)
flat_dependant.dependencies.extend(flat_sub.dependencies)
return flat_dependant
@@ -191,11 +209,21 @@ def get_flat_params(dependant: Dependant) -> List[ModelField]:
return path_params + query_params + header_params + cookie_params
def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
def _get_signature(call: Callable[..., Any]) -> inspect.Signature:
if sys.version_info >= (3, 10):
signature = inspect.signature(call, eval_str=True)
try:
signature = inspect.signature(call, eval_str=True)
except NameError:
# Handle type annotations with if TYPE_CHECKING, not used by FastAPI
# e.g. dependency return types
signature = inspect.signature(call)
else:
signature = inspect.signature(call)
return signature
def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
signature = _get_signature(call)
unwrapped = inspect.unwrap(call)
globalns = getattr(unwrapped, "__globals__", {})
typed_params = [
@@ -221,10 +249,7 @@ def get_typed_annotation(annotation: Any, globalns: Dict[str, Any]) -> Any:
def get_typed_return_annotation(call: Callable[..., Any]) -> Any:
if sys.version_info >= (3, 10):
signature = inspect.signature(call, eval_str=True)
else:
signature = inspect.signature(call)
signature = _get_signature(call)
unwrapped = inspect.unwrap(call)
annotation = signature.return_annotation
@@ -258,11 +283,6 @@ def get_dependant(
path_param_names = get_path_param_names(path)
endpoint_signature = get_typed_signature(call)
signature_params = endpoint_signature.parameters
if isinstance(call, SecurityBase):
security_requirement = SecurityRequirement(
security_scheme=call, scopes=current_scopes
)
dependant.security_requirements.append(security_requirement)
for param_name, param in signature_params.items():
is_path_param = param_name in path_param_names
param_details = analyze_param(

View File

@@ -1,4 +1,4 @@
from typing import Any, Dict, Optional, Sequence, Type, Union
from typing import Any, Dict, Optional, Sequence, Type, TypedDict, Union
from annotated_doc import Doc
from pydantic import BaseModel, create_model
@@ -7,6 +7,13 @@ from starlette.exceptions import WebSocketException as StarletteWebSocketExcepti
from typing_extensions import Annotated
class EndpointContext(TypedDict, total=False):
function: str
path: str
file: str
line: int
class HTTPException(StarletteHTTPException):
"""
An HTTP exception you can raise in your own code to show errors to the client.
@@ -155,30 +162,72 @@ class DependencyScopeError(FastAPIError):
class ValidationException(Exception):
def __init__(self, errors: Sequence[Any]) -> None:
def __init__(
self,
errors: Sequence[Any],
*,
endpoint_ctx: Optional[EndpointContext] = None,
) -> None:
self._errors = errors
self.endpoint_ctx = endpoint_ctx
ctx = endpoint_ctx or {}
self.endpoint_function = ctx.get("function")
self.endpoint_path = ctx.get("path")
self.endpoint_file = ctx.get("file")
self.endpoint_line = ctx.get("line")
def errors(self) -> Sequence[Any]:
return self._errors
def _format_endpoint_context(self) -> str:
if not (self.endpoint_file and self.endpoint_line and self.endpoint_function):
if self.endpoint_path:
return f"\n Endpoint: {self.endpoint_path}"
return ""
context = f'\n File "{self.endpoint_file}", line {self.endpoint_line}, in {self.endpoint_function}'
if self.endpoint_path:
context += f"\n {self.endpoint_path}"
return context
def __str__(self) -> str:
message = f"{len(self._errors)} validation error{'s' if len(self._errors) != 1 else ''}:\n"
for err in self._errors:
message += f" {err}\n"
message += self._format_endpoint_context()
return message.rstrip()
class RequestValidationError(ValidationException):
def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:
super().__init__(errors)
def __init__(
self,
errors: Sequence[Any],
*,
body: Any = None,
endpoint_ctx: Optional[EndpointContext] = None,
) -> None:
super().__init__(errors, endpoint_ctx=endpoint_ctx)
self.body = body
class WebSocketRequestValidationError(ValidationException):
pass
def __init__(
self,
errors: Sequence[Any],
*,
endpoint_ctx: Optional[EndpointContext] = None,
) -> None:
super().__init__(errors, endpoint_ctx=endpoint_ctx)
class ResponseValidationError(ValidationException):
def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:
super().__init__(errors)
def __init__(
self,
errors: Sequence[Any],
*,
body: Any = None,
endpoint_ctx: Optional[EndpointContext] = None,
) -> None:
super().__init__(errors, endpoint_ctx=endpoint_ctx)
self.body = body
def __str__(self) -> str:
message = f"{len(self._errors)} validation errors:\n"
for err in self._errors:
message += f" {err}\n"
return message

View File

@@ -81,18 +81,18 @@ def get_openapi_security_definitions(
security_definitions = {}
# Use a dict to merge scopes for same security scheme
operation_security_dict: Dict[str, List[str]] = {}
for security_requirement in flat_dependant.security_requirements:
for security_dependency in flat_dependant._security_dependencies:
security_definition = jsonable_encoder(
security_requirement.security_scheme.model,
security_dependency._security_scheme.model,
by_alias=True,
exclude_none=True,
)
security_name = security_requirement.security_scheme.scheme_name
security_name = security_dependency._security_scheme.scheme_name
security_definitions[security_name] = security_definition
# Merge scopes for the same security scheme
if security_name not in operation_security_dict:
operation_security_dict[security_name] = []
for scope in security_requirement.scopes or []:
for scope in security_dependency.oauth_scopes or []:
if scope not in operation_security_dict[security_name]:
operation_security_dict[security_name].append(scope)
operation_security = [

View File

@@ -46,6 +46,7 @@ from fastapi.dependencies.utils import (
)
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import (
EndpointContext,
FastAPIError,
RequestValidationError,
ResponseValidationError,
@@ -212,6 +213,33 @@ def _merge_lifespan_context(
return merged_lifespan # type: ignore[return-value]
# Cache for endpoint context to avoid re-extracting on every request
_endpoint_context_cache: Dict[int, EndpointContext] = {}
def _extract_endpoint_context(func: Any) -> EndpointContext:
"""Extract endpoint context with caching to avoid repeated file I/O."""
func_id = id(func)
if func_id in _endpoint_context_cache:
return _endpoint_context_cache[func_id]
try:
ctx: EndpointContext = {}
if (source_file := inspect.getsourcefile(func)) is not None:
ctx["file"] = source_file
if (line_number := inspect.getsourcelines(func)[1]) is not None:
ctx["line"] = line_number
if (func_name := getattr(func, "__name__", None)) is not None:
ctx["function"] = func_name
except Exception:
ctx = EndpointContext()
_endpoint_context_cache[func_id] = ctx
return ctx
async def serialize_response(
*,
field: Optional[ModelField] = None,
@@ -223,6 +251,7 @@ async def serialize_response(
exclude_defaults: bool = False,
exclude_none: bool = False,
is_coroutine: bool = True,
endpoint_ctx: Optional[EndpointContext] = None,
) -> Any:
if field:
errors = []
@@ -245,8 +274,11 @@ async def serialize_response(
elif errors_:
errors.append(errors_)
if errors:
ctx = endpoint_ctx or EndpointContext()
raise ResponseValidationError(
errors=_normalize_errors(errors), body=response_content
errors=_normalize_errors(errors),
body=response_content,
endpoint_ctx=ctx,
)
if hasattr(field, "serialize"):
@@ -318,6 +350,18 @@ def get_request_handler(
"fastapi_middleware_astack not found in request scope"
)
# Extract endpoint context for error messages
endpoint_ctx = (
_extract_endpoint_context(dependant.call)
if dependant.call
else EndpointContext()
)
if dependant.path:
# For mounted sub-apps, include the mount path prefix
mount_path = request.scope.get("root_path", "").rstrip("/")
endpoint_ctx["path"] = f"{request.method} {mount_path}{dependant.path}"
# Read body and auto-close files
try:
body: Any = None
@@ -355,6 +399,7 @@ def get_request_handler(
}
],
body=e.doc,
endpoint_ctx=endpoint_ctx,
)
raise validation_error from e
except HTTPException:
@@ -414,6 +459,7 @@ def get_request_handler(
exclude_defaults=response_model_exclude_defaults,
exclude_none=response_model_exclude_none,
is_coroutine=is_coroutine,
endpoint_ctx=endpoint_ctx,
)
response = actual_response_class(content, **response_args)
if not is_body_allowed_for_status_code(response.status_code):
@@ -421,7 +467,7 @@ def get_request_handler(
response.headers.raw.extend(solved_result.response.headers.raw)
if errors:
validation_error = RequestValidationError(
_normalize_errors(errors), body=body
_normalize_errors(errors), body=body, endpoint_ctx=endpoint_ctx
)
raise validation_error
@@ -438,6 +484,15 @@ def get_websocket_app(
embed_body_fields: bool = False,
) -> Callable[[WebSocket], Coroutine[Any, Any, Any]]:
async def app(websocket: WebSocket) -> None:
endpoint_ctx = (
_extract_endpoint_context(dependant.call)
if dependant.call
else EndpointContext()
)
if dependant.path:
# For mounted sub-apps, include the mount path prefix
mount_path = websocket.scope.get("root_path", "").rstrip("/")
endpoint_ctx["path"] = f"WS {mount_path}{dependant.path}"
async_exit_stack = websocket.scope.get("fastapi_inner_astack")
assert isinstance(async_exit_stack, AsyncExitStack), (
"fastapi_inner_astack not found in request scope"
@@ -451,7 +506,8 @@ def get_websocket_app(
)
if solved_result.errors:
raise WebSocketRequestValidationError(
_normalize_errors(solved_result.errors)
_normalize_errors(solved_result.errors),
endpoint_ctx=endpoint_ctx,
)
assert dependant.call is not None, "dependant.call must be a function"
await dependant.call(**solved_result.values)

View File

@@ -236,8 +236,15 @@ ignore = [
"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"]

View File

@@ -132,7 +132,7 @@ def on_pre_page(page: Page, *, config: MkDocsConfig, files: Files) -> Page:
def on_page_markdown(
markdown: str, *, page: Page, config: MkDocsConfig, files: Files
) -> str:
# Set matadata["social"]["cards_layout_options"]["title"] to clean title (without
# Set metadata["social"]["cards_layout_options"]["title"] to clean title (without
# permalink)
title = page.title
clean_title = title.split("{ #")[0]

View File

@@ -0,0 +1,141 @@
from typing import List
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from typing_extensions import Annotated
from .utils import needs_pydanticv2
@pytest.fixture(name="client")
def get_client():
from pydantic import (
BaseModel,
ConfigDict,
PlainSerializer,
TypeAdapter,
WithJsonSchema,
)
class FakeNumpyArray:
def __init__(self):
self.data = [1.0, 2.0, 3.0]
FakeNumpyArrayPydantic = Annotated[
FakeNumpyArray,
WithJsonSchema(TypeAdapter(List[float]).json_schema()),
PlainSerializer(lambda v: v.data),
]
class MyModel(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
custom_field: FakeNumpyArrayPydantic
app = FastAPI()
@app.get("/")
def test() -> MyModel:
return MyModel(custom_field=FakeNumpyArray())
client = TestClient(app)
return client
@needs_pydanticv2
def test_get(client: TestClient):
response = client.get("/")
assert response.json() == {"custom_field": [1.0, 2.0, 3.0]}
@needs_pydanticv2
def test_typeadapter():
# This test is only to confirm that Pydantic alone is working as expected
from pydantic import (
BaseModel,
ConfigDict,
PlainSerializer,
TypeAdapter,
WithJsonSchema,
)
class FakeNumpyArray:
def __init__(self):
self.data = [1.0, 2.0, 3.0]
FakeNumpyArrayPydantic = Annotated[
FakeNumpyArray,
WithJsonSchema(TypeAdapter(List[float]).json_schema()),
PlainSerializer(lambda v: v.data),
]
class MyModel(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
custom_field: FakeNumpyArrayPydantic
ta = TypeAdapter(MyModel)
assert ta.dump_python(MyModel(custom_field=FakeNumpyArray())) == {
"custom_field": [1.0, 2.0, 3.0]
}
assert ta.json_schema() == snapshot(
{
"properties": {
"custom_field": {
"items": {"type": "number"},
"title": "Custom Field",
"type": "array",
}
},
"required": ["custom_field"],
"title": "MyModel",
"type": "object",
}
)
@needs_pydanticv2
def test_openapi_schema(client: TestClient):
response = client.get("openapi.json")
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/": {
"get": {
"summary": "Test",
"operationId": "test__get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/MyModel"
}
}
},
}
},
}
}
},
"components": {
"schemas": {
"MyModel": {
"properties": {
"custom_field": {
"items": {"type": "number"},
"type": "array",
"title": "Custom Field",
}
},
"type": "object",
"required": ["custom_field"],
"title": "MyModel",
}
}
},
}
)

View File

@@ -48,6 +48,34 @@ async_callable_gen_dependency = AsyncCallableGenDependency()
methods_dependency = MethodsDependency()
@app.get("/callable-dependency-class")
async def get_callable_dependency_class(
value: str, instance: CallableDependency = Depends()
):
return instance(value)
@app.get("/callable-gen-dependency-class")
async def get_callable_gen_dependency_class(
value: str, instance: CallableGenDependency = Depends()
):
return next(instance(value))
@app.get("/async-callable-dependency-class")
async def get_async_callable_dependency_class(
value: str, instance: AsyncCallableDependency = Depends()
):
return await instance(value)
@app.get("/async-callable-gen-dependency-class")
async def get_async_callable_gen_dependency_class(
value: str, instance: AsyncCallableGenDependency = Depends()
):
return await instance(value).__anext__()
@app.get("/callable-dependency")
async def get_callable_dependency(value: str = Depends(callable_dependency)):
return value
@@ -114,6 +142,10 @@ client = TestClient(app)
("/synchronous-method-gen-dependency", "synchronous-method-gen-dependency"),
("/asynchronous-method-dependency", "asynchronous-method-dependency"),
("/asynchronous-method-gen-dependency", "asynchronous-method-gen-dependency"),
("/callable-dependency-class", "callable-dependency-class"),
("/callable-gen-dependency-class", "callable-gen-dependency-class"),
("/async-callable-dependency-class", "async-callable-dependency-class"),
("/async-callable-gen-dependency-class", "async-callable-gen-dependency-class"),
],
)
def test_class_dependency(route, value):

View File

@@ -24,6 +24,18 @@ class Item(BaseModel):
model_config = {"json_schema_serialization_defaults_required": True}
if PYDANTIC_V2:
from pydantic import computed_field
class WithComputedField(BaseModel):
name: str
@computed_field
@property
def computed_field(self) -> str:
return f"computed {self.name}"
def get_app_client(separate_input_output_schemas: bool = True) -> TestClient:
app = FastAPI(separate_input_output_schemas=separate_input_output_schemas)
@@ -46,6 +58,14 @@ def get_app_client(separate_input_output_schemas: bool = True) -> TestClient:
Item(name="Plumbus"),
]
if PYDANTIC_V2:
@app.post("/with-computed-field/")
def create_with_computed_field(
with_computed_field: WithComputedField,
) -> WithComputedField:
return with_computed_field
client = TestClient(app)
return client
@@ -131,6 +151,23 @@ def test_read_items():
)
@needs_pydanticv2
def test_with_computed_field():
client = get_app_client()
client_no = get_app_client(separate_input_output_schemas=False)
response = client.post("/with-computed-field/", json={"name": "example"})
response2 = client_no.post("/with-computed-field/", json={"name": "example"})
assert response.status_code == response2.status_code == 200, response.text
assert (
response.json()
== response2.json()
== {
"name": "example",
"computed_field": "computed example",
}
)
@needs_pydanticv2
def test_openapi_schema():
client = get_app_client()
@@ -245,6 +282,44 @@ def test_openapi_schema():
},
}
},
"/with-computed-field/": {
"post": {
"summary": "Create With Computed Field",
"operationId": "create_with_computed_field_with_computed_field__post",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/WithComputedField-Input"
}
}
},
"required": True,
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/WithComputedField-Output"
}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
},
},
"components": {
"schemas": {
@@ -333,6 +408,25 @@ def test_openapi_schema():
"required": ["subname", "sub_description", "tags"],
"title": "SubItem",
},
"WithComputedField-Input": {
"properties": {"name": {"type": "string", "title": "Name"}},
"type": "object",
"required": ["name"],
"title": "WithComputedField",
},
"WithComputedField-Output": {
"properties": {
"name": {"type": "string", "title": "Name"},
"computed_field": {
"type": "string",
"title": "Computed Field",
"readOnly": True,
},
},
"type": "object",
"required": ["name", "computed_field"],
"title": "WithComputedField",
},
"ValidationError": {
"properties": {
"loc": {
@@ -458,6 +552,44 @@ def test_openapi_schema_no_separate():
},
}
},
"/with-computed-field/": {
"post": {
"summary": "Create With Computed Field",
"operationId": "create_with_computed_field_with_computed_field__post",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/WithComputedField-Input"
}
}
},
"required": True,
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/WithComputedField-Output"
}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
},
},
"components": {
"schemas": {
@@ -508,6 +640,25 @@ def test_openapi_schema_no_separate():
"required": ["subname"],
"title": "SubItem",
},
"WithComputedField-Input": {
"properties": {"name": {"type": "string", "title": "Name"}},
"type": "object",
"required": ["name"],
"title": "WithComputedField",
},
"WithComputedField-Output": {
"properties": {
"name": {"type": "string", "title": "Name"},
"computed_field": {
"type": "string",
"title": "Computed Field",
"readOnly": True,
},
},
"type": "object",
"required": ["name", "computed_field"],
"title": "WithComputedField",
},
"ValidationError": {
"properties": {
"loc": {

View File

@@ -0,0 +1,52 @@
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from typing import List, Union
from dirty_equals import IsUUID
from fastapi import FastAPI
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
@dataclass
class Item:
id: uuid.UUID
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("/item", response_model=Item)
async def read_item():
return {
"id": uuid.uuid4(),
"name": "Island In The Moon",
"price": 12.99,
"description": "A place to be be playin' and havin' fun",
"tags": ["breater"],
}
client = TestClient(app)
def test_annotations():
response = client.get("/item")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"id": IsUUID(),
"name": "Island In The Moon",
"price": 12.99,
"tags": ["breater"],
"description": "A place to be be playin' and havin' fun",
"tax": None,
}
)

View File

@@ -2,10 +2,11 @@
from typing import Optional
from fastapi import APIRouter, FastAPI, Security
from fastapi import APIRouter, Depends, FastAPI, Security
from fastapi.security import OAuth2AuthorizationCodeBearer
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from typing_extensions import Annotated
oauth2_scheme = OAuth2AuthorizationCodeBearer(
authorizationUrl="authorize",
@@ -14,7 +15,12 @@ oauth2_scheme = OAuth2AuthorizationCodeBearer(
scopes={"read": "Read access", "write": "Write access"},
)
app = FastAPI(dependencies=[Security(oauth2_scheme)])
async def get_token(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
return token
app = FastAPI(dependencies=[Depends(get_token)])
@app.get("/")
@@ -22,11 +28,26 @@ async def root():
return {"message": "Hello World"}
@app.get(
"/with-oauth2-scheme",
dependencies=[Security(oauth2_scheme, scopes=["read", "write"])],
)
async def read_with_oauth2_scheme():
return {"message": "Admin Access"}
@app.get(
"/with-get-token", dependencies=[Security(get_token, scopes=["read", "write"])]
)
async def read_with_get_token():
return {"message": "Admin Access"}
router = APIRouter(dependencies=[Security(oauth2_scheme, scopes=["read"])])
@router.get("/items/")
async def read_items(token: Optional[str] = Security(oauth2_scheme)):
async def read_items(token: Optional[str] = Depends(oauth2_scheme)):
return {"token": token}
@@ -48,6 +69,22 @@ def test_root():
assert response.json() == {"message": "Hello World"}
def test_read_with_oauth2_scheme():
response = client.get(
"/with-oauth2-scheme", headers={"Authorization": "Bearer testtoken"}
)
assert response.status_code == 200, response.text
assert response.json() == {"message": "Admin Access"}
def test_read_with_get_token():
response = client.get(
"/with-get-token", headers={"Authorization": "Bearer testtoken"}
)
assert response.status_code == 200, response.text
assert response.json() == {"message": "Admin Access"}
def test_read_token():
response = client.get("/items/", headers={"Authorization": "Bearer testtoken"})
assert response.status_code == 200, response.text
@@ -81,6 +118,36 @@ def test_openapi_schema():
"security": [{"OAuth2AuthorizationCodeBearer": []}],
}
},
"/with-oauth2-scheme": {
"get": {
"summary": "Read With Oauth2 Scheme",
"operationId": "read_with_oauth2_scheme_with_oauth2_scheme_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"security": [
{"OAuth2AuthorizationCodeBearer": ["read", "write"]}
],
}
},
"/with-get-token": {
"get": {
"summary": "Read With Get Token",
"operationId": "read_with_get_token_with_get_token_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"security": [
{"OAuth2AuthorizationCodeBearer": ["read", "write"]}
],
}
},
"/items/": {
"get": {
"summary": "Read Items",

View File

@@ -0,0 +1,79 @@
# Ref: https://github.com/fastapi/fastapi/issues/14454
from fastapi import Depends, FastAPI, Security
from fastapi.security import OAuth2AuthorizationCodeBearer
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from typing_extensions import Annotated
oauth2_scheme = OAuth2AuthorizationCodeBearer(
authorizationUrl="api/oauth/authorize",
tokenUrl="/api/oauth/token",
scopes={"read": "Read access", "write": "Write access"},
)
async def get_token(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
return token
app = FastAPI(dependencies=[Depends(get_token)])
@app.get("/admin", dependencies=[Security(get_token, scopes=["read", "write"])])
async def read_admin():
return {"message": "Admin Access"}
client = TestClient(app)
def test_read_admin():
response = client.get("/admin", headers={"Authorization": "Bearer faketoken"})
assert response.status_code == 200, response.text
assert response.json() == {"message": "Admin Access"}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/admin": {
"get": {
"summary": "Read Admin",
"operationId": "read_admin_admin_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"security": [
{"OAuth2AuthorizationCodeBearer": ["read", "write"]}
],
}
}
},
"components": {
"securitySchemes": {
"OAuth2AuthorizationCodeBearer": {
"type": "oauth2",
"flows": {
"authorizationCode": {
"scopes": {
"read": "Read access",
"write": "Write access",
},
"authorizationUrl": "api/oauth/authorize",
"tokenUrl": "/api/oauth/token",
}
},
}
}
},
}
)

View File

@@ -0,0 +1,80 @@
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from typing_extensions import Annotated
if TYPE_CHECKING: # pragma: no cover
from collections.abc import AsyncGenerator
class DummyClient:
async def get_people(self) -> list:
return ["John Doe", "Jane Doe"]
async def close(self) -> None:
pass
async def get_client() -> AsyncGenerator[DummyClient, None]:
client = DummyClient()
yield client
await client.close()
Client = Annotated[DummyClient, Depends(get_client)]
@pytest.fixture(name="client")
def client_fixture() -> TestClient:
app = FastAPI()
@app.get("/")
async def get_people(client: Client) -> list:
return await client.get_people()
client = TestClient(app)
return client
def test_get(client: TestClient):
response = client.get("/")
assert response.status_code == 200, response.text
assert response.json() == ["John Doe", "Jane Doe"]
def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/": {
"get": {
"summary": "Get People",
"operationId": "get_people__get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"items": {},
"type": "array",
"title": "Response Get People Get",
}
}
},
}
},
}
}
},
}
)

View File

@@ -1,21 +1,36 @@
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
client = TestClient(app)
from tests.utils import needs_py310
def test_path_operation():
@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}")
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
@@ -24,7 +39,7 @@ def test_path_operation_img():
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() == {

View File

@@ -1,21 +1,36 @@
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
client = TestClient(app)
from tests.utils import needs_py310
def test_path_operation():
@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}")
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
@@ -24,7 +39,7 @@ def test_path_operation_img():
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() == {

View File

@@ -1,23 +1,38 @@
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
@@ -30,6 +45,6 @@ def test_gzip_request(compress):
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"}

View File

@@ -1,17 +1,36 @@
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
client = TestClient(app)
from tests.utils import needs_py39, needs_py310
def test_endpoint_works():
@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(mod.app)
return client
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(
{

View File

@@ -1,17 +1,32 @@
import importlib
import pytest
from fastapi.testclient import TestClient
from docs_src.custom_request_and_route.tutorial003 import app
client = TestClient(app)
from tests.utils import needs_py310
def test_get():
@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(mod.app)
return client
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

View File

@@ -1,12 +1,28 @@
import importlib
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
from docs_src.dataclasses.tutorial001 import app
client = TestClient(app)
from tests.utils import needs_py310
def test_post_item():
@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(mod.app)
client.headers.clear()
return client
def test_post_item(client: TestClient):
response = client.post("/items/", json={"name": "Foo", "price": 3})
assert response.status_code == 200
assert response.json() == {
@@ -17,7 +33,7 @@ def test_post_item():
}
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(
@@ -45,7 +61,7 @@ def test_post_invalid_item():
)
def test_openapi_schema():
def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == {

View File

@@ -1,12 +1,29 @@
import importlib
import pytest
from dirty_equals import IsDict, IsOneOf
from fastapi.testclient import TestClient
from docs_src.dataclasses.tutorial002 import app
client = TestClient(app)
from tests.utils import needs_py39, needs_py310
def test_get_item():
@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(mod.app)
client.headers.clear()
return client
def test_get_item(client: TestClient):
response = client.get("/items/next")
assert response.status_code == 200
assert response.json() == {
@@ -18,7 +35,7 @@ def test_get_item():
}
def test_openapi_schema():
def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == {

View File

@@ -1,13 +1,28 @@
import importlib
import pytest
from fastapi.testclient import TestClient
from docs_src.dataclasses.tutorial003 import app
from ...utils import needs_pydanticv1, needs_pydanticv2
client = TestClient(app)
from ...utils import needs_py39, needs_py310, needs_pydanticv1, needs_pydanticv2
def test_post_authors_item():
@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(mod.app)
client.headers.clear()
return client
def test_post_authors_item(client: TestClient):
response = client.post(
"/authors/foo/items/",
json=[{"name": "Bar"}, {"name": "Baz", "description": "Drop the Baz"}],
@@ -22,7 +37,7 @@ def test_post_authors_item():
}
def test_get_authors():
def test_get_authors(client: TestClient):
response = client.get("/authors/")
assert response.status_code == 200
assert response.json() == [
@@ -54,7 +69,7 @@ def test_get_authors():
@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() == {
@@ -191,7 +206,7 @@ def test_openapi_schema():
# 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() == {

View File

@@ -1,12 +1,33 @@
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
client = TestClient(app)
from tests.utils import needs_py310
def test_get():
@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
@pytest.fixture(name="client")
def get_client(mod: ModuleType):
client = TestClient(mod.app)
client.headers.clear()
return client
def test_get(client: TestClient):
response = client.post(
"/invoices/", json={"id": "fooinvoice", "customer": "John", "total": 5.3}
)
@@ -14,12 +35,12 @@ def test_get():
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() == {

View File

@@ -1,13 +1,30 @@
import importlib
import pytest
from fastapi.testclient import TestClient
from docs_src.path_operation_advanced_configuration.tutorial004 import app
from ...utils import needs_pydanticv1, needs_pydanticv2
client = TestClient(app)
from ...utils import needs_py39, needs_py310, needs_pydanticv1, needs_pydanticv2
def test_query_params_str_validations():
@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(mod.app)
client.headers.clear()
return client
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() == {
@@ -20,7 +37,7 @@ def test_query_params_str_validations():
@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() == {
@@ -123,7 +140,7 @@ def test_openapi_schema():
# 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() == {

View File

@@ -1,14 +1,24 @@
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

View File

@@ -1,14 +1,24 @@
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

View File

View File

@@ -0,0 +1,288 @@
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",
},
},
},
}

View File

@@ -1,20 +1,45 @@
import importlib
from types import ModuleType
import pytest
from pytest import MonkeyPatch
from ...utils import needs_pydanticv2
from ...utils import needs_py39, needs_pydanticv2
@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(monkeypatch: MonkeyPatch):
from docs_src.settings.app02 import main
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()

View File

@@ -0,0 +1,59 @@
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,
}

View File

View File

@@ -0,0 +1,112 @@
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",
},
},
},
}

View File

@@ -1,16 +1,45 @@
import importlib
from types import ModuleType
import pytest
from fastapi.testclient import TestClient
from docs_src.websockets.tutorial003 import app, html
client = TestClient(app)
from ...utils import needs_py39
def test_get():
@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)
return client
@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:

View File

@@ -1,50 +0,0 @@
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"

View File

@@ -0,0 +1,168 @@
from fastapi import FastAPI, Request, WebSocket
from fastapi.exceptions import (
RequestValidationError,
ResponseValidationError,
WebSocketRequestValidationError,
)
from fastapi.testclient import TestClient
from pydantic import BaseModel
class Item(BaseModel):
id: int
name: str
class ExceptionCapture:
def __init__(self):
self.exception = None
def capture(self, exc):
self.exception = exc
return exc
app = FastAPI()
sub_app = FastAPI()
captured_exception = ExceptionCapture()
app.mount(path="/sub", app=sub_app)
@app.exception_handler(RequestValidationError)
@sub_app.exception_handler(RequestValidationError)
async def request_validation_handler(request: Request, exc: RequestValidationError):
captured_exception.capture(exc)
raise exc
@app.exception_handler(ResponseValidationError)
@sub_app.exception_handler(ResponseValidationError)
async def response_validation_handler(_: Request, exc: ResponseValidationError):
captured_exception.capture(exc)
raise exc
@app.exception_handler(WebSocketRequestValidationError)
@sub_app.exception_handler(WebSocketRequestValidationError)
async def websocket_validation_handler(
websocket: WebSocket, exc: WebSocketRequestValidationError
):
captured_exception.capture(exc)
raise exc
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"user_id": user_id} # pragma: no cover
@app.get("/items/", response_model=Item)
def get_item():
return {"name": "Widget"}
@sub_app.get("/items/", response_model=Item)
def get_sub_item():
return {"name": "Widget"} # pragma: no cover
@app.websocket("/ws/{item_id}")
async def websocket_endpoint(websocket: WebSocket, item_id: int):
await websocket.accept() # pragma: no cover
await websocket.send_text(f"Item: {item_id}") # pragma: no cover
await websocket.close() # pragma: no cover
@sub_app.websocket("/ws/{item_id}")
async def subapp_websocket_endpoint(websocket: WebSocket, item_id: int):
await websocket.accept() # pragma: no cover
await websocket.send_text(f"Item: {item_id}") # pragma: no cover
await websocket.close() # pragma: no cover
client = TestClient(app)
def test_request_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
client.get("/users/invalid")
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "get_user" in error_str
assert "/users/" in error_str
def test_response_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
client.get("/items/")
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "get_item" in error_str
assert "/items/" in error_str
def test_websocket_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
with client.websocket_connect("/ws/invalid"):
pass # pragma: no cover
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "websocket_endpoint" in error_str
assert "/ws/" in error_str
def test_subapp_request_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
client.get("/sub/items/")
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "get_sub_item" in error_str
assert "/sub/items/" in error_str
def test_subapp_websocket_validation_error_includes_endpoint_context():
captured_exception.exception = None
try:
with client.websocket_connect("/sub/ws/invalid"):
pass # pragma: no cover
except Exception:
pass
assert captured_exception.exception is not None
error_str = str(captured_exception.exception)
assert "subapp_websocket_endpoint" in error_str
assert "/sub/ws/" in error_str
def test_validation_error_with_only_path():
errors = [{"type": "missing", "loc": ("body", "name"), "msg": "Field required"}]
exc = RequestValidationError(errors, endpoint_ctx={"path": "GET /api/test"})
error_str = str(exc)
assert "Endpoint: GET /api/test" in error_str
assert 'File "' not in error_str
def test_validation_error_with_no_context():
errors = [{"type": "missing", "loc": ("body", "name"), "msg": "Field required"}]
exc = RequestValidationError(errors, endpoint_ctx={})
error_str = str(exc)
assert "1 validation error:" in error_str
assert "Endpoint" not in error_str
assert 'File "' not in error_str