Browse Source

Add support for Pydantic models for parameters using `Query`, `Cookie`, `Header` (#12199)

pull/12220/head
Sebastián Ramírez 7 months ago
committed by GitHub
parent
commit
55035f440b
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
  1. BIN
      docs/en/docs/img/tutorial/cookie-param-models/image01.png
  2. BIN
      docs/en/docs/img/tutorial/header-param-models/image01.png
  3. BIN
      docs/en/docs/img/tutorial/query-param-models/image01.png
  4. 154
      docs/en/docs/tutorial/cookie-param-models.md
  5. 184
      docs/en/docs/tutorial/header-param-models.md
  6. 196
      docs/en/docs/tutorial/query-param-models.md
  7. 3
      docs/en/mkdocs.yml
  8. 17
      docs_src/cookie_param_models/tutorial001.py
  9. 18
      docs_src/cookie_param_models/tutorial001_an.py
  10. 17
      docs_src/cookie_param_models/tutorial001_an_py310.py
  11. 17
      docs_src/cookie_param_models/tutorial001_an_py39.py
  12. 15
      docs_src/cookie_param_models/tutorial001_py310.py
  13. 19
      docs_src/cookie_param_models/tutorial002.py
  14. 20
      docs_src/cookie_param_models/tutorial002_an.py
  15. 19
      docs_src/cookie_param_models/tutorial002_an_py310.py
  16. 19
      docs_src/cookie_param_models/tutorial002_an_py39.py
  17. 20
      docs_src/cookie_param_models/tutorial002_pv1.py
  18. 21
      docs_src/cookie_param_models/tutorial002_pv1_an.py
  19. 20
      docs_src/cookie_param_models/tutorial002_pv1_an_py310.py
  20. 20
      docs_src/cookie_param_models/tutorial002_pv1_an_py39.py
  21. 18
      docs_src/cookie_param_models/tutorial002_pv1_py310.py
  22. 17
      docs_src/cookie_param_models/tutorial002_py310.py
  23. 19
      docs_src/header_param_models/tutorial001.py
  24. 20
      docs_src/header_param_models/tutorial001_an.py
  25. 19
      docs_src/header_param_models/tutorial001_an_py310.py
  26. 19
      docs_src/header_param_models/tutorial001_an_py39.py
  27. 17
      docs_src/header_param_models/tutorial001_py310.py
  28. 19
      docs_src/header_param_models/tutorial001_py39.py
  29. 21
      docs_src/header_param_models/tutorial002.py
  30. 22
      docs_src/header_param_models/tutorial002_an.py
  31. 21
      docs_src/header_param_models/tutorial002_an_py310.py
  32. 21
      docs_src/header_param_models/tutorial002_an_py39.py
  33. 22
      docs_src/header_param_models/tutorial002_pv1.py
  34. 23
      docs_src/header_param_models/tutorial002_pv1_an.py
  35. 22
      docs_src/header_param_models/tutorial002_pv1_an_py310.py
  36. 22
      docs_src/header_param_models/tutorial002_pv1_an_py39.py
  37. 20
      docs_src/header_param_models/tutorial002_pv1_py310.py
  38. 22
      docs_src/header_param_models/tutorial002_pv1_py39.py
  39. 19
      docs_src/header_param_models/tutorial002_py310.py
  40. 21
      docs_src/header_param_models/tutorial002_py39.py
  41. 19
      docs_src/query_param_models/tutorial001.py
  42. 19
      docs_src/query_param_models/tutorial001_an.py
  43. 18
      docs_src/query_param_models/tutorial001_an_py310.py
  44. 17
      docs_src/query_param_models/tutorial001_an_py39.py
  45. 18
      docs_src/query_param_models/tutorial001_py310.py
  46. 17
      docs_src/query_param_models/tutorial001_py39.py
  47. 21
      docs_src/query_param_models/tutorial002.py
  48. 21
      docs_src/query_param_models/tutorial002_an.py
  49. 20
      docs_src/query_param_models/tutorial002_an_py310.py
  50. 19
      docs_src/query_param_models/tutorial002_an_py39.py
  51. 22
      docs_src/query_param_models/tutorial002_pv1.py
  52. 22
      docs_src/query_param_models/tutorial002_pv1_an.py
  53. 21
      docs_src/query_param_models/tutorial002_pv1_an_py310.py
  54. 20
      docs_src/query_param_models/tutorial002_pv1_an_py39.py
  55. 21
      docs_src/query_param_models/tutorial002_pv1_py310.py
  56. 20
      docs_src/query_param_models/tutorial002_pv1_py39.py
  57. 20
      docs_src/query_param_models/tutorial002_py310.py
  58. 19
      docs_src/query_param_models/tutorial002_py39.py
  59. 90
      fastapi/dependencies/utils.py
  60. 86
      fastapi/openapi/utils.py
  61. 39
      scripts/playwright/cookie_param_models/image01.py
  62. 38
      scripts/playwright/header_param_models/image01.py
  63. 41
      scripts/playwright/query_param_models/image01.py
  64. 0
      tests/test_tutorial/test_cookie_param_models/__init__.py
  65. 205
      tests/test_tutorial/test_cookie_param_models/test_tutorial001.py
  66. 233
      tests/test_tutorial/test_cookie_param_models/test_tutorial002.py
  67. 0
      tests/test_tutorial/test_header_param_models/__init__.py
  68. 238
      tests/test_tutorial/test_header_param_models/test_tutorial001.py
  69. 249
      tests/test_tutorial/test_header_param_models/test_tutorial002.py
  70. 0
      tests/test_tutorial/test_query_param_models/__init__.py
  71. 260
      tests/test_tutorial/test_query_param_models/test_tutorial001.py
  72. 282
      tests/test_tutorial/test_query_param_models/test_tutorial002.py

BIN
docs/en/docs/img/tutorial/cookie-param-models/image01.png

Binary file not shown.

After

Width:  |  Height:  |  Size: 44 KiB

BIN
docs/en/docs/img/tutorial/header-param-models/image01.png

Binary file not shown.

After

Width:  |  Height:  |  Size: 61 KiB

BIN
docs/en/docs/img/tutorial/query-param-models/image01.png

Binary file not shown.

After

Width:  |  Height:  |  Size: 44 KiB

154
docs/en/docs/tutorial/cookie-param-models.md

@ -0,0 +1,154 @@
# Cookie Parameter Models
If you have a group of **cookies** that are related, you can create a **Pydantic model** to declare them. 🍪
This would allow you to **re-use the model** in **multiple places** and also to declare validations and metadata for all the parameters at once. 😎
/// note
This is supported since FastAPI version `0.115.0`. 🤓
///
/// tip
This same technique applies to `Query`, `Cookie`, and `Header`. 😎
///
## Cookies with a Pydantic Model
Declare the **cookie** parameters that you need in a **Pydantic model**, and then declare the parameter as `Cookie`:
//// tab | Python 3.10+
```Python hl_lines="9-12 16"
{!> ../../../docs_src/cookie_param_models/tutorial001_an_py310.py!}
```
////
//// tab | Python 3.9+
```Python hl_lines="9-12 16"
{!> ../../../docs_src/cookie_param_models/tutorial001_an_py39.py!}
```
////
//// tab | Python 3.8+
```Python hl_lines="10-13 17"
{!> ../../../docs_src/cookie_param_models/tutorial001_an.py!}
```
////
//// tab | Python 3.10+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="7-10 14"
{!> ../../../docs_src/cookie_param_models/tutorial001_py310.py!}
```
////
//// tab | Python 3.8+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="9-12 16"
{!> ../../../docs_src/cookie_param_models/tutorial001.py!}
```
////
**FastAPI** will **extract** the data for **each field** from the **cookies** received in the request and give you the Pydantic model you defined.
## Check the Docs
You can see the defined cookies in the docs UI at `/docs`:
<div class="screenshot">
<img src="/img/tutorial/cookie-param-models/image01.png">
</div>
/// info
Have in mind that, as **browsers handle cookies** in special ways and behind the scenes, they **don't** easily allow **JavaScript** to touch them.
If you go to the **API docs UI** at `/docs` you will be able to see the **documentation** for cookies for your *path operations*.
But even if you **fill the data** and click "Execute", because the docs UI works with **JavaScript**, the cookies won't be sent, and you will see an **error** message as if you didn't write any values.
///
## Forbid Extra Cookies
In some special use cases (probably not very common), you might want to **restrict** the cookies that you want to receive.
Your API now has the power to control its own <abbr title="This is a joke, just in case. It has nothing to do with cookie consents, but it's funny that even the API can now reject the poor cookies. Have a cookie. 🍪">cookie consent</abbr>. 🤪🍪
You can use Pydantic's model configuration to `forbid` any `extra` fields:
//// tab | Python 3.9+
```Python hl_lines="10"
{!> ../../../docs_src/cookie_param_models/tutorial002_an_py39.py!}
```
////
//// tab | Python 3.8+
```Python hl_lines="11"
{!> ../../../docs_src/cookie_param_models/tutorial002_an.py!}
```
////
//// tab | Python 3.8+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="10"
{!> ../../../docs_src/cookie_param_models/tutorial002.py!}
```
////
If a client tries to send some **extra cookies**, they will receive an **error** response.
Poor cookie banners with all their effort to get your consent for the <abbr title="This is another joke. Don't pay attention to me. Have some coffee for your cookie. ☕">API to reject it</abbr>. 🍪
For example, if the client tries to send a `santa_tracker` cookie with a value of `good-list-please`, the client will receive an **error** response telling them that the `santa_tracker` <abbr title="Santa disapproves the lack of cookies. 🎅 Okay, no more cookie jokes.">cookie is not allowed</abbr>:
```json
{
"detail": [
{
"type": "extra_forbidden",
"loc": ["cookie", "santa_tracker"],
"msg": "Extra inputs are not permitted",
"input": "good-list-please",
}
]
}
```
## Summary
You can use **Pydantic models** to declare <abbr title="Have a last cookie before you go. 🍪">**cookies**</abbr> in **FastAPI**. 😎

184
docs/en/docs/tutorial/header-param-models.md

@ -0,0 +1,184 @@
# Header Parameter Models
If you have a group of related **header parameters**, you can create a **Pydantic model** to declare them.
This would allow you to **re-use the model** in **multiple places** and also to declare validations and metadata for all the parameters at once. 😎
/// note
This is supported since FastAPI version `0.115.0`. 🤓
///
## Header Parameters with a Pydantic Model
Declare the **header parameters** that you need in a **Pydantic model**, and then declare the parameter as `Header`:
//// tab | Python 3.10+
```Python hl_lines="9-14 18"
{!> ../../../docs_src/header_param_models/tutorial001_an_py310.py!}
```
////
//// tab | Python 3.9+
```Python hl_lines="9-14 18"
{!> ../../../docs_src/header_param_models/tutorial001_an_py39.py!}
```
////
//// tab | Python 3.8+
```Python hl_lines="10-15 19"
{!> ../../../docs_src/header_param_models/tutorial001_an.py!}
```
////
//// tab | Python 3.10+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="7-12 16"
{!> ../../../docs_src/header_param_models/tutorial001_py310.py!}
```
////
//// tab | Python 3.9+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="9-14 18"
{!> ../../../docs_src/header_param_models/tutorial001_py39.py!}
```
////
//// tab | Python 3.8+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="7-12 16"
{!> ../../../docs_src/header_param_models/tutorial001_py310.py!}
```
////
**FastAPI** will **extract** the data for **each field** from the **headers** in the request and give you the Pydantic model you defined.
## Check the Docs
You can see the required headers in the docs UI at `/docs`:
<div class="screenshot">
<img src="/img/tutorial/header-param-models/image01.png">
</div>
## Forbid Extra Headers
In some special use cases (probably not very common), you might want to **restrict** the headers that you want to receive.
You can use Pydantic's model configuration to `forbid` any `extra` fields:
//// tab | Python 3.10+
```Python hl_lines="10"
{!> ../../../docs_src/header_param_models/tutorial002_an_py310.py!}
```
////
//// tab | Python 3.9+
```Python hl_lines="10"
{!> ../../../docs_src/header_param_models/tutorial002_an_py39.py!}
```
////
//// tab | Python 3.8+
```Python hl_lines="11"
{!> ../../../docs_src/header_param_models/tutorial002_an.py!}
```
////
//// tab | Python 3.10+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="8"
{!> ../../../docs_src/header_param_models/tutorial002_py310.py!}
```
////
//// tab | Python 3.9+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="10"
{!> ../../../docs_src/header_param_models/tutorial002_py39.py!}
```
////
//// tab | Python 3.8+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="10"
{!> ../../../docs_src/header_param_models/tutorial002.py!}
```
////
If a client tries to send some **extra headers**, they will receive an **error** response.
For example, if the client tries to send a `tool` header with a value of `plumbus`, they will receive an **error** response telling them that the header parameter `tool` is not allowed:
```json
{
"detail": [
{
"type": "extra_forbidden",
"loc": ["header", "tool"],
"msg": "Extra inputs are not permitted",
"input": "plumbus",
}
]
}
```
## Summary
You can use **Pydantic models** to declare **headers** in **FastAPI**. 😎

196
docs/en/docs/tutorial/query-param-models.md

@ -0,0 +1,196 @@
# Query Parameter Models
If you have a group of **query parameters** that are related, you can create a **Pydantic model** to declare them.
This would allow you to **re-use the model** in **multiple places** and also to declare validations and metadata for all the parameters at once. 😎
/// note
This is supported since FastAPI version `0.115.0`. 🤓
///
## Query Parameters with a Pydantic Model
Declare the **query parameters** that you need in a **Pydantic model**, and then declare the parameter as `Query`:
//// tab | Python 3.10+
```Python hl_lines="9-13 17"
{!> ../../../docs_src/query_param_models/tutorial001_an_py310.py!}
```
////
//// tab | Python 3.9+
```Python hl_lines="8-12 16"
{!> ../../../docs_src/query_param_models/tutorial001_an_py39.py!}
```
////
//// tab | Python 3.8+
```Python hl_lines="10-14 18"
{!> ../../../docs_src/query_param_models/tutorial001_an.py!}
```
////
//// tab | Python 3.10+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="9-13 17"
{!> ../../../docs_src/query_param_models/tutorial001_py310.py!}
```
////
//// tab | Python 3.9+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="8-12 16"
{!> ../../../docs_src/query_param_models/tutorial001_py39.py!}
```
////
//// tab | Python 3.8+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="9-13 17"
{!> ../../../docs_src/query_param_models/tutorial001_py310.py!}
```
////
**FastAPI** will **extract** the data for **each field** from the **query parameters** in the request and give you the Pydantic model you defined.
## Check the Docs
You can see the query parameters in the docs UI at `/docs`:
<div class="screenshot">
<img src="/img/tutorial/query-param-models/image01.png">
</div>
## Forbid Extra Query Parameters
In some special use cases (probably not very common), you might want to **restrict** the query parameters that you want to receive.
You can use Pydantic's model configuration to `forbid` any `extra` fields:
//// tab | Python 3.10+
```Python hl_lines="10"
{!> ../../../docs_src/query_param_models/tutorial002_an_py310.py!}
```
////
//// tab | Python 3.9+
```Python hl_lines="9"
{!> ../../../docs_src/query_param_models/tutorial002_an_py39.py!}
```
////
//// tab | Python 3.8+
```Python hl_lines="11"
{!> ../../../docs_src/query_param_models/tutorial002_an.py!}
```
////
//// tab | Python 3.10+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="10"
{!> ../../../docs_src/query_param_models/tutorial002_py310.py!}
```
////
//// tab | Python 3.9+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="9"
{!> ../../../docs_src/query_param_models/tutorial002_py39.py!}
```
////
//// tab | Python 3.8+ non-Annotated
/// tip
Prefer to use the `Annotated` version if possible.
///
```Python hl_lines="11"
{!> ../../../docs_src/query_param_models/tutorial002.py!}
```
////
If a client tries to send some **extra** data in the **query parameters**, they will receive an **error** response.
For example, if the client tries to send a `tool` query parameter with a value of `plumbus`, like:
```http
https://example.com/items/?limit=10&tool=plumbus
```
They will receive an **error** response telling them that the query parameter `tool` is not allowed:
```json
{
"detail": [
{
"type": "extra_forbidden",
"loc": ["query", "tool"],
"msg": "Extra inputs are not permitted",
"input": "plumbus"
}
]
}
```
## Summary
You can use **Pydantic models** to declare **query parameters** in **FastAPI**. 😎
/// tip
Spoiler alert: you can also use Pydantic models to declare cookies and headers, but you will read about that later in the tutorial. 🤫
///

3
docs/en/mkdocs.yml

@ -118,6 +118,7 @@ nav:
- tutorial/body.md - tutorial/body.md
- tutorial/query-params-str-validations.md - tutorial/query-params-str-validations.md
- tutorial/path-params-numeric-validations.md - tutorial/path-params-numeric-validations.md
- tutorial/query-param-models.md
- tutorial/body-multiple-params.md - tutorial/body-multiple-params.md
- tutorial/body-fields.md - tutorial/body-fields.md
- tutorial/body-nested-models.md - tutorial/body-nested-models.md
@ -125,6 +126,8 @@ nav:
- tutorial/extra-data-types.md - tutorial/extra-data-types.md
- tutorial/cookie-params.md - tutorial/cookie-params.md
- tutorial/header-params.md - tutorial/header-params.md
- tutorial/cookie-param-models.md
- tutorial/header-param-models.md
- tutorial/response-model.md - tutorial/response-model.md
- tutorial/extra-models.md - tutorial/extra-models.md
- tutorial/response-status-code.md - tutorial/response-status-code.md

17
docs_src/cookie_param_models/tutorial001.py

@ -0,0 +1,17 @@
from typing import Union
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
session_id: str
fatebook_tracker: Union[str, None] = None
googall_tracker: Union[str, None] = None
@app.get("/items/")
async def read_items(cookies: Cookies = Cookie()):
return cookies

18
docs_src/cookie_param_models/tutorial001_an.py

@ -0,0 +1,18 @@
from typing import Union
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class Cookies(BaseModel):
session_id: str
fatebook_tracker: Union[str, None] = None
googall_tracker: Union[str, None] = None
@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

17
docs_src/cookie_param_models/tutorial001_an_py310.py

@ -0,0 +1,17 @@
from typing import Annotated
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
session_id: str
fatebook_tracker: str | None = None
googall_tracker: str | None = None
@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

17
docs_src/cookie_param_models/tutorial001_an_py39.py

@ -0,0 +1,17 @@
from typing import Annotated, Union
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
session_id: str
fatebook_tracker: Union[str, None] = None
googall_tracker: Union[str, None] = None
@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

15
docs_src/cookie_param_models/tutorial001_py310.py

@ -0,0 +1,15 @@
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
session_id: str
fatebook_tracker: str | None = None
googall_tracker: str | None = None
@app.get("/items/")
async def read_items(cookies: Cookies = Cookie()):
return cookies

19
docs_src/cookie_param_models/tutorial002.py

@ -0,0 +1,19 @@
from typing import Union
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
model_config = {"extra": "forbid"}
session_id: str
fatebook_tracker: Union[str, None] = None
googall_tracker: Union[str, None] = None
@app.get("/items/")
async def read_items(cookies: Cookies = Cookie()):
return cookies

20
docs_src/cookie_param_models/tutorial002_an.py

@ -0,0 +1,20 @@
from typing import Union
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class Cookies(BaseModel):
model_config = {"extra": "forbid"}
session_id: str
fatebook_tracker: Union[str, None] = None
googall_tracker: Union[str, None] = None
@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

19
docs_src/cookie_param_models/tutorial002_an_py310.py

@ -0,0 +1,19 @@
from typing import Annotated
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
model_config = {"extra": "forbid"}
session_id: str
fatebook_tracker: str | None = None
googall_tracker: str | None = None
@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

19
docs_src/cookie_param_models/tutorial002_an_py39.py

@ -0,0 +1,19 @@
from typing import Annotated, Union
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
model_config = {"extra": "forbid"}
session_id: str
fatebook_tracker: Union[str, None] = None
googall_tracker: Union[str, None] = None
@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

20
docs_src/cookie_param_models/tutorial002_pv1.py

@ -0,0 +1,20 @@
from typing import Union
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
class Config:
extra = "forbid"
session_id: str
fatebook_tracker: Union[str, None] = None
googall_tracker: Union[str, None] = None
@app.get("/items/")
async def read_items(cookies: Cookies = Cookie()):
return cookies

21
docs_src/cookie_param_models/tutorial002_pv1_an.py

@ -0,0 +1,21 @@
from typing import Union
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class Cookies(BaseModel):
class Config:
extra = "forbid"
session_id: str
fatebook_tracker: Union[str, None] = None
googall_tracker: Union[str, None] = None
@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

20
docs_src/cookie_param_models/tutorial002_pv1_an_py310.py

@ -0,0 +1,20 @@
from typing import Annotated
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
class Config:
extra = "forbid"
session_id: str
fatebook_tracker: str | None = None
googall_tracker: str | None = None
@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

20
docs_src/cookie_param_models/tutorial002_pv1_an_py39.py

@ -0,0 +1,20 @@
from typing import Annotated, Union
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
class Config:
extra = "forbid"
session_id: str
fatebook_tracker: Union[str, None] = None
googall_tracker: Union[str, None] = None
@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
return cookies

18
docs_src/cookie_param_models/tutorial002_pv1_py310.py

@ -0,0 +1,18 @@
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
class Config:
extra = "forbid"
session_id: str
fatebook_tracker: str | None = None
googall_tracker: str | None = None
@app.get("/items/")
async def read_items(cookies: Cookies = Cookie()):
return cookies

17
docs_src/cookie_param_models/tutorial002_py310.py

@ -0,0 +1,17 @@
from fastapi import Cookie, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Cookies(BaseModel):
model_config = {"extra": "forbid"}
session_id: str
fatebook_tracker: str | None = None
googall_tracker: str | None = None
@app.get("/items/")
async def read_items(cookies: Cookies = Cookie()):
return cookies

19
docs_src/header_param_models/tutorial001.py

@ -0,0 +1,19 @@
from typing import List, Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: List[str] = []
@app.get("/items/")
async def read_items(headers: CommonHeaders = Header()):
return headers

20
docs_src/header_param_models/tutorial001_an.py

@ -0,0 +1,20 @@
from typing import List, Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class CommonHeaders(BaseModel):
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: List[str] = []
@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

19
docs_src/header_param_models/tutorial001_an_py310.py

@ -0,0 +1,19 @@
from typing import Annotated
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
host: str
save_data: bool
if_modified_since: str | None = None
traceparent: str | None = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

19
docs_src/header_param_models/tutorial001_an_py39.py

@ -0,0 +1,19 @@
from typing import Annotated, Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

17
docs_src/header_param_models/tutorial001_py310.py

@ -0,0 +1,17 @@
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
host: str
save_data: bool
if_modified_since: str | None = None
traceparent: str | None = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: CommonHeaders = Header()):
return headers

19
docs_src/header_param_models/tutorial001_py39.py

@ -0,0 +1,19 @@
from typing import Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: CommonHeaders = Header()):
return headers

21
docs_src/header_param_models/tutorial002.py

@ -0,0 +1,21 @@
from typing import List, Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
model_config = {"extra": "forbid"}
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: List[str] = []
@app.get("/items/")
async def read_items(headers: CommonHeaders = Header()):
return headers

22
docs_src/header_param_models/tutorial002_an.py

@ -0,0 +1,22 @@
from typing import List, Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class CommonHeaders(BaseModel):
model_config = {"extra": "forbid"}
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: List[str] = []
@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

21
docs_src/header_param_models/tutorial002_an_py310.py

@ -0,0 +1,21 @@
from typing import Annotated
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
model_config = {"extra": "forbid"}
host: str
save_data: bool
if_modified_since: str | None = None
traceparent: str | None = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

21
docs_src/header_param_models/tutorial002_an_py39.py

@ -0,0 +1,21 @@
from typing import Annotated, Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
model_config = {"extra": "forbid"}
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

22
docs_src/header_param_models/tutorial002_pv1.py

@ -0,0 +1,22 @@
from typing import List, Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
class Config:
extra = "forbid"
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: List[str] = []
@app.get("/items/")
async def read_items(headers: CommonHeaders = Header()):
return headers

23
docs_src/header_param_models/tutorial002_pv1_an.py

@ -0,0 +1,23 @@
from typing import List, Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class CommonHeaders(BaseModel):
class Config:
extra = "forbid"
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: List[str] = []
@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

22
docs_src/header_param_models/tutorial002_pv1_an_py310.py

@ -0,0 +1,22 @@
from typing import Annotated
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
class Config:
extra = "forbid"
host: str
save_data: bool
if_modified_since: str | None = None
traceparent: str | None = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

22
docs_src/header_param_models/tutorial002_pv1_an_py39.py

@ -0,0 +1,22 @@
from typing import Annotated, Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
class Config:
extra = "forbid"
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
return headers

20
docs_src/header_param_models/tutorial002_pv1_py310.py

@ -0,0 +1,20 @@
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
class Config:
extra = "forbid"
host: str
save_data: bool
if_modified_since: str | None = None
traceparent: str | None = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: CommonHeaders = Header()):
return headers

22
docs_src/header_param_models/tutorial002_pv1_py39.py

@ -0,0 +1,22 @@
from typing import Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
class Config:
extra = "forbid"
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: CommonHeaders = Header()):
return headers

19
docs_src/header_param_models/tutorial002_py310.py

@ -0,0 +1,19 @@
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
model_config = {"extra": "forbid"}
host: str
save_data: bool
if_modified_since: str | None = None
traceparent: str | None = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: CommonHeaders = Header()):
return headers

21
docs_src/header_param_models/tutorial002_py39.py

@ -0,0 +1,21 @@
from typing import Union
from fastapi import FastAPI, Header
from pydantic import BaseModel
app = FastAPI()
class CommonHeaders(BaseModel):
model_config = {"extra": "forbid"}
host: str
save_data: bool
if_modified_since: Union[str, None] = None
traceparent: Union[str, None] = None
x_tag: list[str] = []
@app.get("/items/")
async def read_items(headers: CommonHeaders = Header()):
return headers

19
docs_src/query_param_models/tutorial001.py

@ -0,0 +1,19 @@
from typing import List
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Literal
app = FastAPI()
class FilterParams(BaseModel):
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: List[str] = []
@app.get("/items/")
async def read_items(filter_query: FilterParams = Query()):
return filter_query

19
docs_src/query_param_models/tutorial001_an.py

@ -0,0 +1,19 @@
from typing import List
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Annotated, Literal
app = FastAPI()
class FilterParams(BaseModel):
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: List[str] = []
@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

18
docs_src/query_param_models/tutorial001_an_py310.py

@ -0,0 +1,18 @@
from typing import Annotated, Literal
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
app = FastAPI()
class FilterParams(BaseModel):
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

17
docs_src/query_param_models/tutorial001_an_py39.py

@ -0,0 +1,17 @@
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Annotated, Literal
app = FastAPI()
class FilterParams(BaseModel):
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

18
docs_src/query_param_models/tutorial001_py310.py

@ -0,0 +1,18 @@
from typing import Literal
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
app = FastAPI()
class FilterParams(BaseModel):
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: FilterParams = Query()):
return filter_query

17
docs_src/query_param_models/tutorial001_py39.py

@ -0,0 +1,17 @@
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Literal
app = FastAPI()
class FilterParams(BaseModel):
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: FilterParams = Query()):
return filter_query

21
docs_src/query_param_models/tutorial002.py

@ -0,0 +1,21 @@
from typing import List
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Literal
app = FastAPI()
class FilterParams(BaseModel):
model_config = {"extra": "forbid"}
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: List[str] = []
@app.get("/items/")
async def read_items(filter_query: FilterParams = Query()):
return filter_query

21
docs_src/query_param_models/tutorial002_an.py

@ -0,0 +1,21 @@
from typing import List
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Annotated, Literal
app = FastAPI()
class FilterParams(BaseModel):
model_config = {"extra": "forbid"}
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: List[str] = []
@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

20
docs_src/query_param_models/tutorial002_an_py310.py

@ -0,0 +1,20 @@
from typing import Annotated, Literal
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
app = FastAPI()
class FilterParams(BaseModel):
model_config = {"extra": "forbid"}
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

19
docs_src/query_param_models/tutorial002_an_py39.py

@ -0,0 +1,19 @@
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Annotated, Literal
app = FastAPI()
class FilterParams(BaseModel):
model_config = {"extra": "forbid"}
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

22
docs_src/query_param_models/tutorial002_pv1.py

@ -0,0 +1,22 @@
from typing import List
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Literal
app = FastAPI()
class FilterParams(BaseModel):
class Config:
extra = "forbid"
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: List[str] = []
@app.get("/items/")
async def read_items(filter_query: FilterParams = Query()):
return filter_query

22
docs_src/query_param_models/tutorial002_pv1_an.py

@ -0,0 +1,22 @@
from typing import List
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Annotated, Literal
app = FastAPI()
class FilterParams(BaseModel):
class Config:
extra = "forbid"
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: List[str] = []
@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

21
docs_src/query_param_models/tutorial002_pv1_an_py310.py

@ -0,0 +1,21 @@
from typing import Annotated, Literal
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
app = FastAPI()
class FilterParams(BaseModel):
class Config:
extra = "forbid"
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

20
docs_src/query_param_models/tutorial002_pv1_an_py39.py

@ -0,0 +1,20 @@
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Annotated, Literal
app = FastAPI()
class FilterParams(BaseModel):
class Config:
extra = "forbid"
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
return filter_query

21
docs_src/query_param_models/tutorial002_pv1_py310.py

@ -0,0 +1,21 @@
from typing import Literal
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
app = FastAPI()
class FilterParams(BaseModel):
class Config:
extra = "forbid"
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: FilterParams = Query()):
return filter_query

20
docs_src/query_param_models/tutorial002_pv1_py39.py

@ -0,0 +1,20 @@
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Literal
app = FastAPI()
class FilterParams(BaseModel):
class Config:
extra = "forbid"
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: FilterParams = Query()):
return filter_query

20
docs_src/query_param_models/tutorial002_py310.py

@ -0,0 +1,20 @@
from typing import Literal
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
app = FastAPI()
class FilterParams(BaseModel):
model_config = {"extra": "forbid"}
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: FilterParams = Query()):
return filter_query

19
docs_src/query_param_models/tutorial002_py39.py

@ -0,0 +1,19 @@
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
from typing_extensions import Literal
app = FastAPI()
class FilterParams(BaseModel):
model_config = {"extra": "forbid"}
limit: int = Field(100, gt=0, le=100)
offset: int = Field(0, ge=0)
order_by: Literal["created_at", "updated_at"] = "created_at"
tags: list[str] = []
@app.get("/items/")
async def read_items(filter_query: FilterParams = Query()):
return filter_query

90
fastapi/dependencies/utils.py

@ -201,14 +201,23 @@ def get_flat_dependant(
return flat_dependant return flat_dependant
def _get_flat_fields_from_params(fields: List[ModelField]) -> List[ModelField]:
if not fields:
return fields
first_field = fields[0]
if len(fields) == 1 and lenient_issubclass(first_field.type_, BaseModel):
fields_to_extract = get_cached_model_fields(first_field.type_)
return fields_to_extract
return fields
def get_flat_params(dependant: Dependant) -> List[ModelField]: def get_flat_params(dependant: Dependant) -> List[ModelField]:
flat_dependant = get_flat_dependant(dependant, skip_repeats=True) flat_dependant = get_flat_dependant(dependant, skip_repeats=True)
return ( path_params = _get_flat_fields_from_params(flat_dependant.path_params)
flat_dependant.path_params query_params = _get_flat_fields_from_params(flat_dependant.query_params)
+ flat_dependant.query_params header_params = _get_flat_fields_from_params(flat_dependant.header_params)
+ flat_dependant.header_params cookie_params = _get_flat_fields_from_params(flat_dependant.cookie_params)
+ flat_dependant.cookie_params return path_params + query_params + header_params + cookie_params
)
def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature: def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
@ -479,7 +488,15 @@ def analyze_param(
field=field field=field
), "Path params must be of one of the supported types" ), "Path params must be of one of the supported types"
elif isinstance(field_info, params.Query): elif isinstance(field_info, params.Query):
assert is_scalar_field(field) or is_scalar_sequence_field(field) assert (
is_scalar_field(field)
or is_scalar_sequence_field(field)
or (
lenient_issubclass(field.type_, BaseModel)
# For Pydantic v1
and getattr(field, "shape", 1) == 1
)
)
return ParamDetails(type_annotation=type_annotation, depends=depends, field=field) return ParamDetails(type_annotation=type_annotation, depends=depends, field=field)
@ -686,11 +703,14 @@ def _validate_value_with_model_field(
return v_, [] return v_, []
def _get_multidict_value(field: ModelField, values: Mapping[str, Any]) -> Any: def _get_multidict_value(
field: ModelField, values: Mapping[str, Any], alias: Union[str, None] = None
) -> Any:
alias = alias or field.alias
if is_sequence_field(field) and isinstance(values, (ImmutableMultiDict, Headers)): if is_sequence_field(field) and isinstance(values, (ImmutableMultiDict, Headers)):
value = values.getlist(field.alias) value = values.getlist(alias)
else: else:
value = values.get(field.alias, None) value = values.get(alias, None)
if ( if (
value is None value is None
or ( or (
@ -712,7 +732,55 @@ def request_params_to_args(
received_params: Union[Mapping[str, Any], QueryParams, Headers], received_params: Union[Mapping[str, Any], QueryParams, Headers],
) -> Tuple[Dict[str, Any], List[Any]]: ) -> Tuple[Dict[str, Any], List[Any]]:
values: Dict[str, Any] = {} values: Dict[str, Any] = {}
errors = [] errors: List[Dict[str, Any]] = []
if not fields:
return values, errors
first_field = fields[0]
fields_to_extract = fields
single_not_embedded_field = False
if len(fields) == 1 and lenient_issubclass(first_field.type_, BaseModel):
fields_to_extract = get_cached_model_fields(first_field.type_)
single_not_embedded_field = True
params_to_process: Dict[str, Any] = {}
processed_keys = set()
for field in fields_to_extract:
alias = None
if isinstance(received_params, Headers):
# Handle fields extracted from a Pydantic Model for a header, each field
# doesn't have a FieldInfo of type Header with the default convert_underscores=True
convert_underscores = getattr(field.field_info, "convert_underscores", True)
if convert_underscores:
alias = (
field.alias
if field.alias != field.name
else field.name.replace("_", "-")
)
value = _get_multidict_value(field, received_params, alias=alias)
if value is not None:
params_to_process[field.name] = value
processed_keys.add(alias or field.alias)
processed_keys.add(field.name)
for key, value in received_params.items():
if key not in processed_keys:
params_to_process[key] = value
if single_not_embedded_field:
field_info = first_field.field_info
assert isinstance(
field_info, params.Param
), "Params must be subclasses of Param"
loc: Tuple[str, ...] = (field_info.in_.value,)
v_, errors_ = _validate_value_with_model_field(
field=first_field, value=params_to_process, values=values, loc=loc
)
return {first_field.name: v_}, errors_
for field in fields: for field in fields:
value = _get_multidict_value(field, received_params) value = _get_multidict_value(field, received_params)
field_info = field.field_info field_info = field.field_info

86
fastapi/openapi/utils.py

@ -16,11 +16,15 @@ from fastapi._compat import (
) )
from fastapi.datastructures import DefaultPlaceholder from fastapi.datastructures import DefaultPlaceholder
from fastapi.dependencies.models import Dependant from fastapi.dependencies.models import Dependant
from fastapi.dependencies.utils import get_flat_dependant, get_flat_params from fastapi.dependencies.utils import (
_get_flat_fields_from_params,
get_flat_dependant,
get_flat_params,
)
from fastapi.encoders import jsonable_encoder from fastapi.encoders import jsonable_encoder
from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX, REF_TEMPLATE from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX, REF_TEMPLATE
from fastapi.openapi.models import OpenAPI from fastapi.openapi.models import OpenAPI
from fastapi.params import Body, Param from fastapi.params import Body, ParamTypes
from fastapi.responses import Response from fastapi.responses import Response
from fastapi.types import ModelNameMap from fastapi.types import ModelNameMap
from fastapi.utils import ( from fastapi.utils import (
@ -87,9 +91,9 @@ def get_openapi_security_definitions(
return security_definitions, operation_security return security_definitions, operation_security
def get_openapi_operation_parameters( def _get_openapi_operation_parameters(
*, *,
all_route_params: Sequence[ModelField], dependant: Dependant,
schema_generator: GenerateJsonSchema, schema_generator: GenerateJsonSchema,
model_name_map: ModelNameMap, model_name_map: ModelNameMap,
field_mapping: Dict[ field_mapping: Dict[
@ -98,33 +102,47 @@ def get_openapi_operation_parameters(
separate_input_output_schemas: bool = True, separate_input_output_schemas: bool = True,
) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]:
parameters = [] parameters = []
for param in all_route_params: flat_dependant = get_flat_dependant(dependant, skip_repeats=True)
field_info = param.field_info path_params = _get_flat_fields_from_params(flat_dependant.path_params)
field_info = cast(Param, field_info) query_params = _get_flat_fields_from_params(flat_dependant.query_params)
if not field_info.include_in_schema: header_params = _get_flat_fields_from_params(flat_dependant.header_params)
continue cookie_params = _get_flat_fields_from_params(flat_dependant.cookie_params)
param_schema = get_schema_from_model_field( parameter_groups = [
field=param, (ParamTypes.path, path_params),
schema_generator=schema_generator, (ParamTypes.query, query_params),
model_name_map=model_name_map, (ParamTypes.header, header_params),
field_mapping=field_mapping, (ParamTypes.cookie, cookie_params),
separate_input_output_schemas=separate_input_output_schemas, ]
) for param_type, param_group in parameter_groups:
parameter = { for param in param_group:
"name": param.alias, field_info = param.field_info
"in": field_info.in_.value, # field_info = cast(Param, field_info)
"required": param.required, if not getattr(field_info, "include_in_schema", True):
"schema": param_schema, continue
} param_schema = get_schema_from_model_field(
if field_info.description: field=param,
parameter["description"] = field_info.description schema_generator=schema_generator,
if field_info.openapi_examples: model_name_map=model_name_map,
parameter["examples"] = jsonable_encoder(field_info.openapi_examples) field_mapping=field_mapping,
elif field_info.example != Undefined: separate_input_output_schemas=separate_input_output_schemas,
parameter["example"] = jsonable_encoder(field_info.example) )
if field_info.deprecated: parameter = {
parameter["deprecated"] = True "name": param.alias,
parameters.append(parameter) "in": param_type.value,
"required": param.required,
"schema": param_schema,
}
if field_info.description:
parameter["description"] = field_info.description
openapi_examples = getattr(field_info, "openapi_examples", None)
example = getattr(field_info, "example", None)
if openapi_examples:
parameter["examples"] = jsonable_encoder(openapi_examples)
elif example != Undefined:
parameter["example"] = jsonable_encoder(example)
if getattr(field_info, "deprecated", None):
parameter["deprecated"] = True
parameters.append(parameter)
return parameters return parameters
@ -247,9 +265,8 @@ def get_openapi_path(
operation.setdefault("security", []).extend(operation_security) operation.setdefault("security", []).extend(operation_security)
if security_definitions: if security_definitions:
security_schemes.update(security_definitions) security_schemes.update(security_definitions)
all_route_params = get_flat_params(route.dependant) operation_parameters = _get_openapi_operation_parameters(
operation_parameters = get_openapi_operation_parameters( dependant=route.dependant,
all_route_params=all_route_params,
schema_generator=schema_generator, schema_generator=schema_generator,
model_name_map=model_name_map, model_name_map=model_name_map,
field_mapping=field_mapping, field_mapping=field_mapping,
@ -379,6 +396,7 @@ def get_openapi_path(
deep_dict_update(openapi_response, process_response) deep_dict_update(openapi_response, process_response)
openapi_response["description"] = description openapi_response["description"] = description
http422 = str(HTTP_422_UNPROCESSABLE_ENTITY) http422 = str(HTTP_422_UNPROCESSABLE_ENTITY)
all_route_params = get_flat_params(route.dependant)
if (all_route_params or route.body_field) and not any( if (all_route_params or route.body_field) and not any(
status in operation["responses"] status in operation["responses"]
for status in [http422, "4XX", "default"] for status in [http422, "4XX", "default"]

39
scripts/playwright/cookie_param_models/image01.py

@ -0,0 +1,39 @@
import subprocess
import time
import httpx
from playwright.sync_api import Playwright, sync_playwright
# Run playwright codegen to generate the code below, copy paste the sections in run()
def run(playwright: Playwright) -> None:
browser = playwright.chromium.launch(headless=False)
# Update the viewport manually
context = browser.new_context(viewport={"width": 960, "height": 1080})
browser = playwright.chromium.launch(headless=False)
context = browser.new_context()
page = context.new_page()
page.goto("http://localhost:8000/docs")
page.get_by_role("link", name="/items/").click()
# Manually add the screenshot
page.screenshot(path="docs/en/docs/img/tutorial/cookie-param-models/image01.png")
# ---------------------
context.close()
browser.close()
process = subprocess.Popen(
["fastapi", "run", "docs_src/cookie_param_models/tutorial001.py"]
)
try:
for _ in range(3):
try:
response = httpx.get("http://localhost:8000/docs")
except httpx.ConnectError:
time.sleep(1)
break
with sync_playwright() as playwright:
run(playwright)
finally:
process.terminate()

38
scripts/playwright/header_param_models/image01.py

@ -0,0 +1,38 @@
import subprocess
import time
import httpx
from playwright.sync_api import Playwright, sync_playwright
# Run playwright codegen to generate the code below, copy paste the sections in run()
def run(playwright: Playwright) -> None:
browser = playwright.chromium.launch(headless=False)
# Update the viewport manually
context = browser.new_context(viewport={"width": 960, "height": 1080})
page = context.new_page()
page.goto("http://localhost:8000/docs")
page.get_by_role("button", name="GET /items/ Read Items").click()
page.get_by_role("button", name="Try it out").click()
# Manually add the screenshot
page.screenshot(path="docs/en/docs/img/tutorial/header-param-models/image01.png")
# ---------------------
context.close()
browser.close()
process = subprocess.Popen(
["fastapi", "run", "docs_src/header_param_models/tutorial001.py"]
)
try:
for _ in range(3):
try:
response = httpx.get("http://localhost:8000/docs")
except httpx.ConnectError:
time.sleep(1)
break
with sync_playwright() as playwright:
run(playwright)
finally:
process.terminate()

41
scripts/playwright/query_param_models/image01.py

@ -0,0 +1,41 @@
import subprocess
import time
import httpx
from playwright.sync_api import Playwright, sync_playwright
# Run playwright codegen to generate the code below, copy paste the sections in run()
def run(playwright: Playwright) -> None:
browser = playwright.chromium.launch(headless=False)
# Update the viewport manually
context = browser.new_context(viewport={"width": 960, "height": 1080})
browser = playwright.chromium.launch(headless=False)
context = browser.new_context()
page = context.new_page()
page.goto("http://localhost:8000/docs")
page.get_by_role("button", name="GET /items/ Read Items").click()
page.get_by_role("button", name="Try it out").click()
page.get_by_role("heading", name="Servers").click()
# Manually add the screenshot
page.screenshot(path="docs/en/docs/img/tutorial/query-param-models/image01.png")
# ---------------------
context.close()
browser.close()
process = subprocess.Popen(
["fastapi", "run", "docs_src/query_param_models/tutorial001.py"]
)
try:
for _ in range(3):
try:
response = httpx.get("http://localhost:8000/docs")
except httpx.ConnectError:
time.sleep(1)
break
with sync_playwright() as playwright:
run(playwright)
finally:
process.terminate()

0
tests/test_tutorial/test_cookie_param_models/__init__.py

205
tests/test_tutorial/test_cookie_param_models/test_tutorial001.py

@ -0,0 +1,205 @@
import importlib
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from tests.utils import needs_py39, needs_py310
@pytest.fixture(
name="client",
params=[
"tutorial001",
pytest.param("tutorial001_py310", marks=needs_py310),
"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.cookie_param_models.{request.param}")
client = TestClient(mod.app)
return client
def test_cookie_param_model(client: TestClient):
with client as c:
c.cookies.set("session_id", "123")
c.cookies.set("fatebook_tracker", "456")
c.cookies.set("googall_tracker", "789")
response = c.get("/items/")
assert response.status_code == 200
assert response.json() == {
"session_id": "123",
"fatebook_tracker": "456",
"googall_tracker": "789",
}
def test_cookie_param_model_defaults(client: TestClient):
with client as c:
c.cookies.set("session_id", "123")
response = c.get("/items/")
assert response.status_code == 200
assert response.json() == {
"session_id": "123",
"fatebook_tracker": None,
"googall_tracker": None,
}
def test_cookie_param_model_invalid(client: TestClient):
response = client.get("/items/")
assert response.status_code == 422
assert response.json() == snapshot(
IsDict(
{
"detail": [
{
"type": "missing",
"loc": ["cookie", "session_id"],
"msg": "Field required",
"input": {},
}
]
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"detail": [
{
"type": "value_error.missing",
"loc": ["cookie", "session_id"],
"msg": "field required",
}
]
}
)
)
def test_cookie_param_model_extra(client: TestClient):
with client as c:
c.cookies.set("session_id", "123")
c.cookies.set("extra", "track-me-here-too")
response = c.get("/items/")
assert response.status_code == 200
assert response.json() == snapshot(
{"session_id": "123", "fatebook_tracker": None, "googall_tracker": None}
)
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": {
"/items/": {
"get": {
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"name": "session_id",
"in": "cookie",
"required": True,
"schema": {"type": "string", "title": "Session Id"},
},
{
"name": "fatebook_tracker",
"in": "cookie",
"required": False,
"schema": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "Fatebook Tracker",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "string",
"title": "Fatebook Tracker",
}
),
},
{
"name": "googall_tracker",
"in": "cookie",
"required": False,
"schema": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "Googall Tracker",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "string",
"title": "Googall Tracker",
}
),
},
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [{"type": "string"}, {"type": "integer"}]
},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
)

233
tests/test_tutorial/test_cookie_param_models/test_tutorial002.py

@ -0,0 +1,233 @@
import importlib
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from tests.utils import needs_py39, needs_py310, needs_pydanticv1, needs_pydanticv2
@pytest.fixture(
name="client",
params=[
pytest.param("tutorial002", marks=needs_pydanticv2),
pytest.param("tutorial002_py310", marks=[needs_py310, needs_pydanticv2]),
pytest.param("tutorial002_an", marks=needs_pydanticv2),
pytest.param("tutorial002_an_py39", marks=[needs_py39, needs_pydanticv2]),
pytest.param("tutorial002_an_py310", marks=[needs_py310, needs_pydanticv2]),
pytest.param("tutorial002_pv1", marks=[needs_pydanticv1, needs_pydanticv1]),
pytest.param("tutorial002_pv1_py310", marks=[needs_py310, needs_pydanticv1]),
pytest.param("tutorial002_pv1_an", marks=[needs_pydanticv1]),
pytest.param("tutorial002_pv1_an_py39", marks=[needs_py39, needs_pydanticv1]),
pytest.param("tutorial002_pv1_an_py310", marks=[needs_py310, needs_pydanticv1]),
],
)
def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.cookie_param_models.{request.param}")
client = TestClient(mod.app)
return client
def test_cookie_param_model(client: TestClient):
with client as c:
c.cookies.set("session_id", "123")
c.cookies.set("fatebook_tracker", "456")
c.cookies.set("googall_tracker", "789")
response = c.get("/items/")
assert response.status_code == 200
assert response.json() == {
"session_id": "123",
"fatebook_tracker": "456",
"googall_tracker": "789",
}
def test_cookie_param_model_defaults(client: TestClient):
with client as c:
c.cookies.set("session_id", "123")
response = c.get("/items/")
assert response.status_code == 200
assert response.json() == {
"session_id": "123",
"fatebook_tracker": None,
"googall_tracker": None,
}
def test_cookie_param_model_invalid(client: TestClient):
response = client.get("/items/")
assert response.status_code == 422
assert response.json() == snapshot(
IsDict(
{
"detail": [
{
"type": "missing",
"loc": ["cookie", "session_id"],
"msg": "Field required",
"input": {},
}
]
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"detail": [
{
"type": "value_error.missing",
"loc": ["cookie", "session_id"],
"msg": "field required",
}
]
}
)
)
def test_cookie_param_model_extra(client: TestClient):
with client as c:
c.cookies.set("session_id", "123")
c.cookies.set("extra", "track-me-here-too")
response = c.get("/items/")
assert response.status_code == 422
assert response.json() == snapshot(
IsDict(
{
"detail": [
{
"type": "extra_forbidden",
"loc": ["cookie", "extra"],
"msg": "Extra inputs are not permitted",
"input": "track-me-here-too",
}
]
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"detail": [
{
"type": "value_error.extra",
"loc": ["cookie", "extra"],
"msg": "extra fields not permitted",
}
]
}
)
)
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": {
"/items/": {
"get": {
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"name": "session_id",
"in": "cookie",
"required": True,
"schema": {"type": "string", "title": "Session Id"},
},
{
"name": "fatebook_tracker",
"in": "cookie",
"required": False,
"schema": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "Fatebook Tracker",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "string",
"title": "Fatebook Tracker",
}
),
},
{
"name": "googall_tracker",
"in": "cookie",
"required": False,
"schema": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "Googall Tracker",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "string",
"title": "Googall Tracker",
}
),
},
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [{"type": "string"}, {"type": "integer"}]
},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
)

0
tests/test_tutorial/test_header_param_models/__init__.py

238
tests/test_tutorial/test_header_param_models/test_tutorial001.py

@ -0,0 +1,238 @@
import importlib
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from tests.utils import needs_py39, needs_py310
@pytest.fixture(
name="client",
params=[
"tutorial001",
pytest.param("tutorial001_py39", marks=needs_py39),
pytest.param("tutorial001_py310", marks=needs_py310),
"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.header_param_models.{request.param}")
client = TestClient(mod.app)
return client
def test_header_param_model(client: TestClient):
response = client.get(
"/items/",
headers=[
("save-data", "true"),
("if-modified-since", "yesterday"),
("traceparent", "123"),
("x-tag", "one"),
("x-tag", "two"),
],
)
assert response.status_code == 200
assert response.json() == {
"host": "testserver",
"save_data": True,
"if_modified_since": "yesterday",
"traceparent": "123",
"x_tag": ["one", "two"],
}
def test_header_param_model_defaults(client: TestClient):
response = client.get("/items/", headers=[("save-data", "true")])
assert response.status_code == 200
assert response.json() == {
"host": "testserver",
"save_data": True,
"if_modified_since": None,
"traceparent": None,
"x_tag": [],
}
def test_header_param_model_invalid(client: TestClient):
response = client.get("/items/")
assert response.status_code == 422
assert response.json() == snapshot(
{
"detail": [
IsDict(
{
"type": "missing",
"loc": ["header", "save_data"],
"msg": "Field required",
"input": {
"x_tag": [],
"host": "testserver",
"accept": "*/*",
"accept-encoding": "gzip, deflate",
"connection": "keep-alive",
"user-agent": "testclient",
},
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "value_error.missing",
"loc": ["header", "save_data"],
"msg": "field required",
}
)
]
}
)
def test_header_param_model_extra(client: TestClient):
response = client.get(
"/items/", headers=[("save-data", "true"), ("tool", "plumbus")]
)
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"host": "testserver",
"save_data": True,
"if_modified_since": None,
"traceparent": None,
"x_tag": [],
}
)
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": {
"/items/": {
"get": {
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"name": "host",
"in": "header",
"required": True,
"schema": {"type": "string", "title": "Host"},
},
{
"name": "save_data",
"in": "header",
"required": True,
"schema": {"type": "boolean", "title": "Save Data"},
},
{
"name": "if_modified_since",
"in": "header",
"required": False,
"schema": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "If Modified Since",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "string",
"title": "If Modified Since",
}
),
},
{
"name": "traceparent",
"in": "header",
"required": False,
"schema": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "Traceparent",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "string",
"title": "Traceparent",
}
),
},
{
"name": "x_tag",
"in": "header",
"required": False,
"schema": {
"type": "array",
"items": {"type": "string"},
"default": [],
"title": "X Tag",
},
},
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [{"type": "string"}, {"type": "integer"}]
},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
)

249
tests/test_tutorial/test_header_param_models/test_tutorial002.py

@ -0,0 +1,249 @@
import importlib
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from tests.utils import needs_py39, needs_py310, needs_pydanticv1, needs_pydanticv2
@pytest.fixture(
name="client",
params=[
pytest.param("tutorial002", marks=needs_pydanticv2),
pytest.param("tutorial002_py310", marks=[needs_py310, needs_pydanticv2]),
pytest.param("tutorial002_an", marks=needs_pydanticv2),
pytest.param("tutorial002_an_py39", marks=[needs_py39, needs_pydanticv2]),
pytest.param("tutorial002_an_py310", marks=[needs_py310, needs_pydanticv2]),
pytest.param("tutorial002_pv1", marks=[needs_pydanticv1, needs_pydanticv1]),
pytest.param("tutorial002_pv1_py310", marks=[needs_py310, needs_pydanticv1]),
pytest.param("tutorial002_pv1_an", marks=[needs_pydanticv1]),
pytest.param("tutorial002_pv1_an_py39", marks=[needs_py39, needs_pydanticv1]),
pytest.param("tutorial002_pv1_an_py310", marks=[needs_py310, needs_pydanticv1]),
],
)
def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.header_param_models.{request.param}")
client = TestClient(mod.app)
client.headers.clear()
return client
def test_header_param_model(client: TestClient):
response = client.get(
"/items/",
headers=[
("save-data", "true"),
("if-modified-since", "yesterday"),
("traceparent", "123"),
("x-tag", "one"),
("x-tag", "two"),
],
)
assert response.status_code == 200, response.text
assert response.json() == {
"host": "testserver",
"save_data": True,
"if_modified_since": "yesterday",
"traceparent": "123",
"x_tag": ["one", "two"],
}
def test_header_param_model_defaults(client: TestClient):
response = client.get("/items/", headers=[("save-data", "true")])
assert response.status_code == 200
assert response.json() == {
"host": "testserver",
"save_data": True,
"if_modified_since": None,
"traceparent": None,
"x_tag": [],
}
def test_header_param_model_invalid(client: TestClient):
response = client.get("/items/")
assert response.status_code == 422
assert response.json() == snapshot(
{
"detail": [
IsDict(
{
"type": "missing",
"loc": ["header", "save_data"],
"msg": "Field required",
"input": {"x_tag": [], "host": "testserver"},
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "value_error.missing",
"loc": ["header", "save_data"],
"msg": "field required",
}
)
]
}
)
def test_header_param_model_extra(client: TestClient):
response = client.get(
"/items/", headers=[("save-data", "true"), ("tool", "plumbus")]
)
assert response.status_code == 422, response.text
assert response.json() == snapshot(
{
"detail": [
IsDict(
{
"type": "extra_forbidden",
"loc": ["header", "tool"],
"msg": "Extra inputs are not permitted",
"input": "plumbus",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "value_error.extra",
"loc": ["header", "tool"],
"msg": "extra fields not permitted",
}
)
]
}
)
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": {
"/items/": {
"get": {
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"name": "host",
"in": "header",
"required": True,
"schema": {"type": "string", "title": "Host"},
},
{
"name": "save_data",
"in": "header",
"required": True,
"schema": {"type": "boolean", "title": "Save Data"},
},
{
"name": "if_modified_since",
"in": "header",
"required": False,
"schema": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "If Modified Since",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "string",
"title": "If Modified Since",
}
),
},
{
"name": "traceparent",
"in": "header",
"required": False,
"schema": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "Traceparent",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "string",
"title": "Traceparent",
}
),
},
{
"name": "x_tag",
"in": "header",
"required": False,
"schema": {
"type": "array",
"items": {"type": "string"},
"default": [],
"title": "X Tag",
},
},
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [{"type": "string"}, {"type": "integer"}]
},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
)

0
tests/test_tutorial/test_query_param_models/__init__.py

260
tests/test_tutorial/test_query_param_models/test_tutorial001.py

@ -0,0 +1,260 @@
import importlib
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from tests.utils import needs_py39, needs_py310
@pytest.fixture(
name="client",
params=[
"tutorial001",
pytest.param("tutorial001_py39", marks=needs_py39),
pytest.param("tutorial001_py310", marks=needs_py310),
"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.query_param_models.{request.param}")
client = TestClient(mod.app)
return client
def test_query_param_model(client: TestClient):
response = client.get(
"/items/",
params={
"limit": 10,
"offset": 5,
"order_by": "updated_at",
"tags": ["tag1", "tag2"],
},
)
assert response.status_code == 200
assert response.json() == {
"limit": 10,
"offset": 5,
"order_by": "updated_at",
"tags": ["tag1", "tag2"],
}
def test_query_param_model_defaults(client: TestClient):
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == {
"limit": 100,
"offset": 0,
"order_by": "created_at",
"tags": [],
}
def test_query_param_model_invalid(client: TestClient):
response = client.get(
"/items/",
params={
"limit": 150,
"offset": -1,
"order_by": "invalid",
},
)
assert response.status_code == 422
assert response.json() == snapshot(
IsDict(
{
"detail": [
{
"type": "less_than_equal",
"loc": ["query", "limit"],
"msg": "Input should be less than or equal to 100",
"input": "150",
"ctx": {"le": 100},
},
{
"type": "greater_than_equal",
"loc": ["query", "offset"],
"msg": "Input should be greater than or equal to 0",
"input": "-1",
"ctx": {"ge": 0},
},
{
"type": "literal_error",
"loc": ["query", "order_by"],
"msg": "Input should be 'created_at' or 'updated_at'",
"input": "invalid",
"ctx": {"expected": "'created_at' or 'updated_at'"},
},
]
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"detail": [
{
"type": "value_error.number.not_le",
"loc": ["query", "limit"],
"msg": "ensure this value is less than or equal to 100",
"ctx": {"limit_value": 100},
},
{
"type": "value_error.number.not_ge",
"loc": ["query", "offset"],
"msg": "ensure this value is greater than or equal to 0",
"ctx": {"limit_value": 0},
},
{
"type": "value_error.const",
"loc": ["query", "order_by"],
"msg": "unexpected value; permitted: 'created_at', 'updated_at'",
"ctx": {
"given": "invalid",
"permitted": ["created_at", "updated_at"],
},
},
]
}
)
)
def test_query_param_model_extra(client: TestClient):
response = client.get(
"/items/",
params={
"limit": 10,
"offset": 5,
"order_by": "updated_at",
"tags": ["tag1", "tag2"],
"tool": "plumbus",
},
)
assert response.status_code == 200
assert response.json() == {
"limit": 10,
"offset": 5,
"order_by": "updated_at",
"tags": ["tag1", "tag2"],
}
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": {
"/items/": {
"get": {
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"name": "limit",
"in": "query",
"required": False,
"schema": {
"type": "integer",
"maximum": 100,
"exclusiveMinimum": 0,
"default": 100,
"title": "Limit",
},
},
{
"name": "offset",
"in": "query",
"required": False,
"schema": {
"type": "integer",
"minimum": 0,
"default": 0,
"title": "Offset",
},
},
{
"name": "order_by",
"in": "query",
"required": False,
"schema": {
"enum": ["created_at", "updated_at"],
"type": "string",
"default": "created_at",
"title": "Order By",
},
},
{
"name": "tags",
"in": "query",
"required": False,
"schema": {
"type": "array",
"items": {"type": "string"},
"default": [],
"title": "Tags",
},
},
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [{"type": "string"}, {"type": "integer"}]
},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
)

282
tests/test_tutorial/test_query_param_models/test_tutorial002.py

@ -0,0 +1,282 @@
import importlib
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
from tests.utils import needs_py39, needs_py310, needs_pydanticv1, needs_pydanticv2
@pytest.fixture(
name="client",
params=[
pytest.param("tutorial002", marks=needs_pydanticv2),
pytest.param("tutorial002_py39", marks=[needs_py39, needs_pydanticv2]),
pytest.param("tutorial002_py310", marks=[needs_py310, needs_pydanticv2]),
pytest.param("tutorial002_an", marks=needs_pydanticv2),
pytest.param("tutorial002_an_py39", marks=[needs_py39, needs_pydanticv2]),
pytest.param("tutorial002_an_py310", marks=[needs_py310, needs_pydanticv2]),
pytest.param("tutorial002_pv1", marks=[needs_pydanticv1, needs_pydanticv1]),
pytest.param("tutorial002_pv1_py39", marks=[needs_py39, needs_pydanticv1]),
pytest.param("tutorial002_pv1_py310", marks=[needs_py310, needs_pydanticv1]),
pytest.param("tutorial002_pv1_an", marks=[needs_pydanticv1]),
pytest.param("tutorial002_pv1_an_py39", marks=[needs_py39, needs_pydanticv1]),
pytest.param("tutorial002_pv1_an_py310", marks=[needs_py310, needs_pydanticv1]),
],
)
def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.query_param_models.{request.param}")
client = TestClient(mod.app)
return client
def test_query_param_model(client: TestClient):
response = client.get(
"/items/",
params={
"limit": 10,
"offset": 5,
"order_by": "updated_at",
"tags": ["tag1", "tag2"],
},
)
assert response.status_code == 200
assert response.json() == {
"limit": 10,
"offset": 5,
"order_by": "updated_at",
"tags": ["tag1", "tag2"],
}
def test_query_param_model_defaults(client: TestClient):
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == {
"limit": 100,
"offset": 0,
"order_by": "created_at",
"tags": [],
}
def test_query_param_model_invalid(client: TestClient):
response = client.get(
"/items/",
params={
"limit": 150,
"offset": -1,
"order_by": "invalid",
},
)
assert response.status_code == 422
assert response.json() == snapshot(
IsDict(
{
"detail": [
{
"type": "less_than_equal",
"loc": ["query", "limit"],
"msg": "Input should be less than or equal to 100",
"input": "150",
"ctx": {"le": 100},
},
{
"type": "greater_than_equal",
"loc": ["query", "offset"],
"msg": "Input should be greater than or equal to 0",
"input": "-1",
"ctx": {"ge": 0},
},
{
"type": "literal_error",
"loc": ["query", "order_by"],
"msg": "Input should be 'created_at' or 'updated_at'",
"input": "invalid",
"ctx": {"expected": "'created_at' or 'updated_at'"},
},
]
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"detail": [
{
"type": "value_error.number.not_le",
"loc": ["query", "limit"],
"msg": "ensure this value is less than or equal to 100",
"ctx": {"limit_value": 100},
},
{
"type": "value_error.number.not_ge",
"loc": ["query", "offset"],
"msg": "ensure this value is greater than or equal to 0",
"ctx": {"limit_value": 0},
},
{
"type": "value_error.const",
"loc": ["query", "order_by"],
"msg": "unexpected value; permitted: 'created_at', 'updated_at'",
"ctx": {
"given": "invalid",
"permitted": ["created_at", "updated_at"],
},
},
]
}
)
)
def test_query_param_model_extra(client: TestClient):
response = client.get(
"/items/",
params={
"limit": 10,
"offset": 5,
"order_by": "updated_at",
"tags": ["tag1", "tag2"],
"tool": "plumbus",
},
)
assert response.status_code == 422
assert response.json() == snapshot(
{
"detail": [
IsDict(
{
"type": "extra_forbidden",
"loc": ["query", "tool"],
"msg": "Extra inputs are not permitted",
"input": "plumbus",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{
"type": "value_error.extra",
"loc": ["query", "tool"],
"msg": "extra fields not permitted",
}
)
]
}
)
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": {
"/items/": {
"get": {
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"name": "limit",
"in": "query",
"required": False,
"schema": {
"type": "integer",
"maximum": 100,
"exclusiveMinimum": 0,
"default": 100,
"title": "Limit",
},
},
{
"name": "offset",
"in": "query",
"required": False,
"schema": {
"type": "integer",
"minimum": 0,
"default": 0,
"title": "Offset",
},
},
{
"name": "order_by",
"in": "query",
"required": False,
"schema": {
"enum": ["created_at", "updated_at"],
"type": "string",
"default": "created_at",
"title": "Order By",
},
},
{
"name": "tags",
"in": "query",
"required": False,
"schema": {
"type": "array",
"items": {"type": "string"},
"default": [],
"title": "Tags",
},
},
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [{"type": "string"}, {"type": "integer"}]
},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
)
Loading…
Cancel
Save