Browse Source

Add support for PEP-593 `Annotated` for specifying dependencies and parameters (#4871)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Sebastián Ramírez <[email protected]>
pull/9269/head
Nadav Zingerman 2 years ago
committed by GitHub
parent
commit
375513f114
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
  1. 18
      docs_src/annotated/tutorial001.py
  2. 17
      docs_src/annotated/tutorial001_py39.py
  3. 21
      docs_src/annotated/tutorial002.py
  4. 20
      docs_src/annotated/tutorial002_py39.py
  5. 15
      docs_src/annotated/tutorial003.py
  6. 16
      docs_src/annotated/tutorial003_py39.py
  7. 258
      fastapi/dependencies/utils.py
  8. 2
      fastapi/param_functions.py
  9. 9
      fastapi/params.py
  10. 9
      fastapi/utils.py
  11. 7
      tests/main.py
  12. 66
      tests/test_ambiguous_params.py
  13. 226
      tests/test_annotated.py
  14. 30
      tests/test_application.py
  15. 5
      tests/test_params_repr.py
  16. 1
      tests/test_path.py
  17. 0
      tests/test_tutorial/test_annotated/__init__.py
  18. 100
      tests/test_tutorial/test_annotated/test_tutorial001.py
  19. 107
      tests/test_tutorial/test_annotated/test_tutorial001_py39.py
  20. 100
      tests/test_tutorial/test_annotated/test_tutorial002.py
  21. 107
      tests/test_tutorial/test_annotated/test_tutorial002_py39.py
  22. 138
      tests/test_tutorial/test_annotated/test_tutorial003.py
  23. 145
      tests/test_tutorial/test_annotated/test_tutorial003_py39.py
  24. 0
      tests/test_tutorial/test_dataclasses/__init__.py

18
docs_src/annotated/tutorial001.py

@ -0,0 +1,18 @@
from typing import Optional
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
CommonParamsDepends = Annotated[dict, Depends(common_parameters)]
@app.get("/items/")
async def read_items(commons: CommonParamsDepends):
return commons

17
docs_src/annotated/tutorial001_py39.py

@ -0,0 +1,17 @@
from typing import Annotated, Optional
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
CommonParamsDepends = Annotated[dict, Depends(common_parameters)]
@app.get("/items/")
async def read_items(commons: CommonParamsDepends):
return commons

21
docs_src/annotated/tutorial002.py

@ -0,0 +1,21 @@
from typing import Optional
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
class CommonQueryParams:
def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
CommonQueryParamsDepends = Annotated[CommonQueryParams, Depends()]
@app.get("/items/")
async def read_items(commons: CommonQueryParamsDepends):
return commons

20
docs_src/annotated/tutorial002_py39.py

@ -0,0 +1,20 @@
from typing import Annotated, Optional
from fastapi import Depends, FastAPI
app = FastAPI()
class CommonQueryParams:
def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
CommonQueryParamsDepends = Annotated[CommonQueryParams, Depends()]
@app.get("/items/")
async def read_items(commons: CommonQueryParamsDepends):
return commons

15
docs_src/annotated/tutorial003.py

@ -0,0 +1,15 @@
from fastapi import FastAPI, Path
from fastapi.param_functions import Query
from typing_extensions import Annotated
app = FastAPI()
@app.get("/items/{item_id}")
async def read_items(item_id: Annotated[int, Path(gt=0)]):
return {"item_id": item_id}
@app.get("/users")
async def read_users(user_id: Annotated[str, Query(min_length=1)] = "me"):
return {"user_id": user_id}

16
docs_src/annotated/tutorial003_py39.py

@ -0,0 +1,16 @@
from typing import Annotated
from fastapi import FastAPI, Path
from fastapi.param_functions import Query
app = FastAPI()
@app.get("/items/{item_id}")
async def read_items(item_id: Annotated[int, Path(gt=0)]):
return {"item_id": item_id}
@app.get("/users")
async def read_users(user_id: Annotated[str, Query(min_length=1)] = "me"):
return {"user_id": user_id}

258
fastapi/dependencies/utils.py

@ -48,7 +48,7 @@ from pydantic.fields import (
Undefined,
)
from pydantic.schema import get_annotation_from_field_info
from pydantic.typing import evaluate_forwardref
from pydantic.typing import evaluate_forwardref, get_args, get_origin
from pydantic.utils import lenient_issubclass
from starlette.background import BackgroundTasks
from starlette.concurrency import run_in_threadpool
@ -56,6 +56,7 @@ from starlette.datastructures import FormData, Headers, QueryParams, UploadFile
from starlette.requests import HTTPConnection, Request
from starlette.responses import Response
from starlette.websockets import WebSocket
from typing_extensions import Annotated
sequence_shapes = {
SHAPE_LIST,
@ -112,18 +113,18 @@ def check_file_field(field: ModelField) -> None:
def get_param_sub_dependant(
*, param: inspect.Parameter, path: str, security_scopes: Optional[List[str]] = None
*,
param_name: str,
depends: params.Depends,
path: str,
security_scopes: Optional[List[str]] = None,
) -> Dependant:
depends: params.Depends = param.default
if depends.dependency:
dependency = depends.dependency
else:
dependency = param.annotation
assert depends.dependency
return get_sub_dependant(
depends=depends,
dependency=dependency,
dependency=depends.dependency,
path=path,
name=param.name,
name=param_name,
security_scopes=security_scopes,
)
@ -298,122 +299,199 @@ def get_dependant(
use_cache=use_cache,
)
for param_name, param in signature_params.items():
if isinstance(param.default, params.Depends):
is_path_param = param_name in path_param_names
type_annotation, depends, param_field = analyze_param(
param_name=param_name,
annotation=param.annotation,
value=param.default,
is_path_param=is_path_param,
)
if depends is not None:
sub_dependant = get_param_sub_dependant(
param=param, path=path, security_scopes=security_scopes
param_name=param_name,
depends=depends,
path=path,
security_scopes=security_scopes,
)
dependant.dependencies.append(sub_dependant)
continue
if add_non_field_param_to_dependency(param=param, dependant=dependant):
continue
param_field = get_param_field(
param=param, default_field_info=params.Query, param_name=param_name
)
if param_name in path_param_names:
assert is_scalar_field(
field=param_field
), "Path params must be of one of the supported types"
ignore_default = not isinstance(param.default, params.Path)
param_field = get_param_field(
param=param,
if add_non_field_param_to_dependency(
param_name=param_name,
default_field_info=params.Path,
force_type=params.ParamTypes.path,
ignore_default=ignore_default,
)
add_param_to_fields(field=param_field, dependant=dependant)
elif is_scalar_field(field=param_field):
add_param_to_fields(field=param_field, dependant=dependant)
elif isinstance(
param.default, (params.Query, params.Header)
) and is_scalar_sequence_field(param_field):
add_param_to_fields(field=param_field, dependant=dependant)
else:
field_info = param_field.field_info
assert isinstance(
field_info, params.Body
), f"Param: {param_field.name} can only be a request body, using Body()"
type_annotation=type_annotation,
dependant=dependant,
):
assert (
param_field is None
), f"Cannot specify multiple FastAPI annotations for {param_name!r}"
continue
assert param_field is not None
if is_body_param(param_field=param_field, is_path_param=is_path_param):
dependant.body_params.append(param_field)
else:
add_param_to_fields(field=param_field, dependant=dependant)
return dependant
def add_non_field_param_to_dependency(
*, param: inspect.Parameter, dependant: Dependant
*, param_name: str, type_annotation: Any, dependant: Dependant
) -> Optional[bool]:
if lenient_issubclass(param.annotation, Request):
dependant.request_param_name = param.name
if lenient_issubclass(type_annotation, Request):
dependant.request_param_name = param_name
return True
elif lenient_issubclass(param.annotation, WebSocket):
dependant.websocket_param_name = param.name
elif lenient_issubclass(type_annotation, WebSocket):
dependant.websocket_param_name = param_name
return True
elif lenient_issubclass(param.annotation, HTTPConnection):
dependant.http_connection_param_name = param.name
elif lenient_issubclass(type_annotation, HTTPConnection):
dependant.http_connection_param_name = param_name
return True
elif lenient_issubclass(param.annotation, Response):
dependant.response_param_name = param.name
elif lenient_issubclass(type_annotation, Response):
dependant.response_param_name = param_name
return True
elif lenient_issubclass(param.annotation, BackgroundTasks):
dependant.background_tasks_param_name = param.name
elif lenient_issubclass(type_annotation, BackgroundTasks):
dependant.background_tasks_param_name = param_name
return True
elif lenient_issubclass(param.annotation, SecurityScopes):
dependant.security_scopes_param_name = param.name
elif lenient_issubclass(type_annotation, SecurityScopes):
dependant.security_scopes_param_name = param_name
return True
return None
def get_param_field(
def analyze_param(
*,
param: inspect.Parameter,
param_name: str,
default_field_info: Type[params.Param] = params.Param,
force_type: Optional[params.ParamTypes] = None,
ignore_default: bool = False,
) -> ModelField:
default_value: Any = Undefined
had_schema = False
if not param.default == param.empty and ignore_default is False:
default_value = param.default
if isinstance(default_value, FieldInfo):
had_schema = True
field_info = default_value
default_value = field_info.default
annotation: Any,
value: Any,
is_path_param: bool,
) -> Tuple[Any, Optional[params.Depends], Optional[ModelField]]:
field_info = None
used_default_field_info = False
depends = None
type_annotation: Any = Any
if (
annotation is not inspect.Signature.empty
and get_origin(annotation) is Annotated # type: ignore[comparison-overlap]
):
annotated_args = get_args(annotation)
type_annotation = annotated_args[0]
fastapi_annotations = [
arg
for arg in annotated_args[1:]
if isinstance(arg, (FieldInfo, params.Depends))
]
assert (
len(fastapi_annotations) <= 1
), f"Cannot specify multiple `Annotated` FastAPI arguments for {param_name!r}"
fastapi_annotation = next(iter(fastapi_annotations), None)
if isinstance(fastapi_annotation, FieldInfo):
field_info = fastapi_annotation
assert field_info.default is Undefined or field_info.default is Required, (
f"`{field_info.__class__.__name__}` default value cannot be set in"
f" `Annotated` for {param_name!r}. Set the default value with `=` instead."
)
if value is not inspect.Signature.empty:
assert not is_path_param, "Path parameters cannot have default values"
field_info.default = value
else:
field_info.default = Required
elif isinstance(fastapi_annotation, params.Depends):
depends = fastapi_annotation
elif annotation is not inspect.Signature.empty:
type_annotation = annotation
if isinstance(value, params.Depends):
assert depends is None, (
"Cannot specify `Depends` in `Annotated` and default value"
f" together for {param_name!r}"
)
assert field_info is None, (
"Cannot specify a FastAPI annotation in `Annotated` and `Depends` as a"
f" default value together for {param_name!r}"
)
depends = value
elif isinstance(value, FieldInfo):
assert field_info is None, (
"Cannot specify FastAPI annotations in `Annotated` and default value"
f" together for {param_name!r}"
)
field_info = value
if depends is not None and depends.dependency is None:
depends.dependency = type_annotation
if lenient_issubclass(
type_annotation,
(Request, WebSocket, HTTPConnection, Response, BackgroundTasks, SecurityScopes),
):
assert depends is None, f"Cannot specify `Depends` for type {type_annotation!r}"
assert (
field_info is None
), f"Cannot specify FastAPI annotation for type {type_annotation!r}"
elif field_info is None and depends is None:
default_value = value if value is not inspect.Signature.empty else Required
if is_path_param:
# We might check here that `default_value is Required`, but the fact is that the same
# parameter might sometimes be a path parameter and sometimes not. See
# `tests/test_infer_param_optionality.py` for an example.
field_info = params.Path()
else:
field_info = params.Query(default=default_value)
used_default_field_info = True
field = None
if field_info is not None:
if is_path_param:
assert isinstance(field_info, params.Path), (
f"Cannot use `{field_info.__class__.__name__}` for path param"
f" {param_name!r}"
)
elif (
isinstance(field_info, params.Param)
and getattr(field_info, "in_", None) is None
):
field_info.in_ = default_field_info.in_
if force_type:
field_info.in_ = force_type # type: ignore
else:
field_info = default_field_info(default=default_value)
required = True
if default_value is Required or ignore_default:
required = True
default_value = None
elif default_value is not Undefined:
required = False
annotation: Any = Any
if not param.annotation == param.empty:
annotation = param.annotation
annotation = get_annotation_from_field_info(annotation, field_info, param_name)
field_info.in_ = params.ParamTypes.query
annotation = get_annotation_from_field_info(
annotation if annotation is not inspect.Signature.empty else Any,
field_info,
param_name,
)
if not field_info.alias and getattr(field_info, "convert_underscores", None):
alias = param.name.replace("_", "-")
alias = param_name.replace("_", "-")
else:
alias = field_info.alias or param.name
alias = field_info.alias or param_name
field = create_response_field(
name=param.name,
name=param_name,
type_=annotation,
default=default_value,
default=field_info.default,
alias=alias,
required=required,
required=field_info.default in (Required, Undefined),
field_info=field_info,
)
if not had_schema and not is_scalar_field(field=field):
field.field_info = params.Body(field_info.default)
if not had_schema and lenient_issubclass(field.type_, UploadFile):
if used_default_field_info:
if lenient_issubclass(field.type_, UploadFile):
field.field_info = params.File(field_info.default)
elif not is_scalar_field(field=field):
field.field_info = params.Body(field_info.default)
return type_annotation, depends, field
return field
def is_body_param(*, param_field: ModelField, is_path_param: bool) -> bool:
if is_path_param:
assert is_scalar_field(
field=param_field
), "Path params must be of one of the supported types"
return False
elif is_scalar_field(field=param_field):
return False
elif isinstance(
param_field.field_info, (params.Query, params.Header)
) and is_scalar_sequence_field(param_field):
return False
else:
assert isinstance(
param_field.field_info, params.Body
), f"Param: {param_field.name} can only be a request body, using Body()"
return True
def add_param_to_fields(*, field: ModelField, dependant: Dependant) -> None:

2
fastapi/param_functions.py

@ -5,7 +5,7 @@ from pydantic.fields import Undefined
def Path( # noqa: N802
default: Any = Undefined,
default: Any = ...,
*,
alias: Optional[str] = None,
title: Optional[str] = None,

9
fastapi/params.py

@ -62,7 +62,7 @@ class Path(Param):
def __init__(
self,
default: Any = Undefined,
default: Any = ...,
*,
alias: Optional[str] = None,
title: Optional[str] = None,
@ -80,9 +80,10 @@ class Path(Param):
include_in_schema: bool = True,
**extra: Any,
):
assert default is ..., "Path parameters cannot have a default value"
self.in_ = self.in_
super().__init__(
default=...,
default=default,
alias=alias,
title=title,
description=description,
@ -279,7 +280,7 @@ class Body(FieldInfo):
class Form(Body):
def __init__(
self,
default: Any,
default: Any = Undefined,
*,
media_type: str = "application/x-www-form-urlencoded",
alias: Optional[str] = None,
@ -319,7 +320,7 @@ class Form(Body):
class File(Form):
def __init__(
self,
default: Any,
default: Any = Undefined,
*,
media_type: str = "multipart/form-data",
alias: Optional[str] = None,

9
fastapi/utils.py

@ -1,4 +1,3 @@
import functools
import re
import warnings
from dataclasses import is_dataclass
@ -73,8 +72,8 @@ def create_response_field(
class_validators = class_validators or {}
field_info = field_info or FieldInfo()
response_field = functools.partial(
ModelField,
try:
return ModelField(
name=name,
type_=type_,
class_validators=class_validators,
@ -82,10 +81,8 @@ def create_response_field(
required=required,
model_config=model_config,
alias=alias,
field_info=field_info,
)
try:
return response_field(field_info=field_info)
except RuntimeError:
raise fastapi.exceptions.FastAPIError(
"Invalid args for response field! Hint: "

7
tests/main.py

@ -49,12 +49,7 @@ def get_bool_id(item_id: bool):
@app.get("/path/param/{item_id}")
def get_path_param_id(item_id: str = Path()):
return item_id
@app.get("/path/param-required/{item_id}")
def get_path_param_required_id(item_id: str = Path()):
def get_path_param_id(item_id: Optional[str] = Path()):
return item_id

66
tests/test_ambiguous_params.py

@ -0,0 +1,66 @@
import pytest
from fastapi import Depends, FastAPI, Path
from fastapi.param_functions import Query
from typing_extensions import Annotated
app = FastAPI()
def test_no_annotated_defaults():
with pytest.raises(
AssertionError, match="Path parameters cannot have a default value"
):
@app.get("/items/{item_id}/")
async def get_item(item_id: Annotated[int, Path(default=1)]):
pass # pragma: nocover
with pytest.raises(
AssertionError,
match=(
"`Query` default value cannot be set in `Annotated` for 'item_id'. Set the"
" default value with `=` instead."
),
):
@app.get("/")
async def get(item_id: Annotated[int, Query(default=1)]):
pass # pragma: nocover
def test_no_multiple_annotations():
async def dep():
pass # pragma: nocover
with pytest.raises(
AssertionError,
match="Cannot specify multiple `Annotated` FastAPI arguments for 'foo'",
):
@app.get("/")
async def get(foo: Annotated[int, Query(min_length=1), Query()]):
pass # pragma: nocover
with pytest.raises(
AssertionError,
match=(
"Cannot specify `Depends` in `Annotated` and default value"
" together for 'foo'"
),
):
@app.get("/")
async def get2(foo: Annotated[int, Depends(dep)] = Depends(dep)):
pass # pragma: nocover
with pytest.raises(
AssertionError,
match=(
"Cannot specify a FastAPI annotation in `Annotated` and `Depends` as a"
" default value together for 'foo'"
),
):
@app.get("/")
async def get3(foo: Annotated[int, Query(min_length=1)] = Depends(dep)):
pass # pragma: nocover

226
tests/test_annotated.py

@ -0,0 +1,226 @@
import pytest
from fastapi import FastAPI, Query
from fastapi.testclient import TestClient
from typing_extensions import Annotated
app = FastAPI()
@app.get("/default")
async def default(foo: Annotated[str, Query()] = "foo"):
return {"foo": foo}
@app.get("/required")
async def required(foo: Annotated[str, Query(min_length=1)]):
return {"foo": foo}
@app.get("/multiple")
async def multiple(foo: Annotated[str, object(), Query(min_length=1)]):
return {"foo": foo}
@app.get("/unrelated")
async def unrelated(foo: Annotated[str, object()]):
return {"foo": foo}
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/default": {
"get": {
"summary": "Default",
"operationId": "default_default_get",
"parameters": [
{
"required": False,
"schema": {"title": "Foo", "type": "string", "default": "foo"},
"name": "foo",
"in": "query",
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
},
"/required": {
"get": {
"summary": "Required",
"operationId": "required_required_get",
"parameters": [
{
"required": True,
"schema": {"title": "Foo", "minLength": 1, "type": "string"},
"name": "foo",
"in": "query",
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
},
"/multiple": {
"get": {
"summary": "Multiple",
"operationId": "multiple_multiple_get",
"parameters": [
{
"required": True,
"schema": {"title": "Foo", "minLength": 1, "type": "string"},
"name": "foo",
"in": "query",
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
},
"/unrelated": {
"get": {
"summary": "Unrelated",
"operationId": "unrelated_unrelated_get",
"parameters": [
{
"required": True,
"schema": {"title": "Foo", "type": "string"},
"name": "foo",
"in": "query",
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
},
},
"components": {
"schemas": {
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
}
},
}
foo_is_missing = {
"detail": [
{
"loc": ["query", "foo"],
"msg": "field required",
"type": "value_error.missing",
}
]
}
foo_is_short = {
"detail": [
{
"ctx": {"limit_value": 1},
"loc": ["query", "foo"],
"msg": "ensure this value has at least 1 characters",
"type": "value_error.any_str.min_length",
}
]
}
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[
("/default", 200, {"foo": "foo"}),
("/default?foo=bar", 200, {"foo": "bar"}),
("/required?foo=bar", 200, {"foo": "bar"}),
("/required", 422, foo_is_missing),
("/required?foo=", 422, foo_is_short),
("/multiple?foo=bar", 200, {"foo": "bar"}),
("/multiple", 422, foo_is_missing),
("/multiple?foo=", 422, foo_is_short),
("/unrelated?foo=bar", 200, {"foo": "bar"}),
("/unrelated", 422, foo_is_missing),
("/openapi.json", 200, openapi_schema),
],
)
def test_get(path, expected_status, expected_response):
response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response

30
tests/test_application.py

@ -225,36 +225,6 @@ openapi_schema = {
],
}
},
"/path/param-required/{item_id}": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Get Path Param Required Id",
"operationId": "get_path_param_required_id_path_param_required__item_id__get",
"parameters": [
{
"required": True,
"schema": {"title": "Item Id", "type": "string"},
"name": "item_id",
"in": "path",
}
],
}
},
"/path/param-minlength/{item_id}": {
"get": {
"responses": {

5
tests/test_params_repr.py

@ -19,8 +19,9 @@ def test_param_repr(params):
assert repr(Param(params)) == "Param(" + str(params) + ")"
def test_path_repr(params):
assert repr(Path(params)) == "Path(Ellipsis)"
def test_path_repr():
assert repr(Path()) == "Path(Ellipsis)"
assert repr(Path(...)) == "Path(Ellipsis)"
def test_query_repr(params):

1
tests/test_path.py

@ -193,7 +193,6 @@ response_less_than_equal_3 = {
("/path/bool/False", 200, False),
("/path/bool/false", 200, False),
("/path/param/foo", 200, "foo"),
("/path/param-required/foo", 200, "foo"),
("/path/param-minlength/foo", 200, "foo"),
("/path/param-minlength/fo", 422, response_at_least_3),
("/path/param-maxlength/foo", 200, "foo"),

0
tests/test_tutorial/test_annotated/__init__.py

100
tests/test_tutorial/test_annotated/test_tutorial001.py

@ -0,0 +1,100 @@
import pytest
from fastapi.testclient import TestClient
from docs_src.annotated.tutorial001 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/items/": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"required": False,
"schema": {"title": "Q", "type": "string"},
"name": "q",
"in": "query",
},
{
"required": False,
"schema": {"title": "Skip", "type": "integer", "default": 0},
"name": "skip",
"in": "query",
},
{
"required": False,
"schema": {"title": "Limit", "type": "integer", "default": 100},
"name": "limit",
"in": "query",
},
],
}
},
},
"components": {
"schemas": {
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
}
},
}
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[
("/items", 200, {"q": None, "skip": 0, "limit": 100}),
("/items?q=foo", 200, {"q": "foo", "skip": 0, "limit": 100}),
("/items?q=foo&skip=5", 200, {"q": "foo", "skip": 5, "limit": 100}),
("/items?q=foo&skip=5&limit=30", 200, {"q": "foo", "skip": 5, "limit": 30}),
("/openapi.json", 200, openapi_schema),
],
)
def test_get(path, expected_status, expected_response):
response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response

107
tests/test_tutorial/test_annotated/test_tutorial001_py39.py

@ -0,0 +1,107 @@
import pytest
from fastapi.testclient import TestClient
from ...utils import needs_py39
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/items/": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"required": False,
"schema": {"title": "Q", "type": "string"},
"name": "q",
"in": "query",
},
{
"required": False,
"schema": {"title": "Skip", "type": "integer", "default": 0},
"name": "skip",
"in": "query",
},
{
"required": False,
"schema": {"title": "Limit", "type": "integer", "default": 100},
"name": "limit",
"in": "query",
},
],
}
},
},
"components": {
"schemas": {
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
}
},
}
@pytest.fixture(name="client")
def get_client():
from docs_src.annotated.tutorial001_py39 import app
client = TestClient(app)
return client
@needs_py39
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[
("/items", 200, {"q": None, "skip": 0, "limit": 100}),
("/items?q=foo", 200, {"q": "foo", "skip": 0, "limit": 100}),
("/items?q=foo&skip=5", 200, {"q": "foo", "skip": 5, "limit": 100}),
("/items?q=foo&skip=5&limit=30", 200, {"q": "foo", "skip": 5, "limit": 30}),
("/openapi.json", 200, openapi_schema),
],
)
def test_get(path, expected_status, expected_response, client):
response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response

100
tests/test_tutorial/test_annotated/test_tutorial002.py

@ -0,0 +1,100 @@
import pytest
from fastapi.testclient import TestClient
from docs_src.annotated.tutorial002 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/items/": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"required": False,
"schema": {"title": "Q", "type": "string"},
"name": "q",
"in": "query",
},
{
"required": False,
"schema": {"title": "Skip", "type": "integer", "default": 0},
"name": "skip",
"in": "query",
},
{
"required": False,
"schema": {"title": "Limit", "type": "integer", "default": 100},
"name": "limit",
"in": "query",
},
],
}
},
},
"components": {
"schemas": {
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
}
},
}
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[
("/items", 200, {"q": None, "skip": 0, "limit": 100}),
("/items?q=foo", 200, {"q": "foo", "skip": 0, "limit": 100}),
("/items?q=foo&skip=5", 200, {"q": "foo", "skip": 5, "limit": 100}),
("/items?q=foo&skip=5&limit=30", 200, {"q": "foo", "skip": 5, "limit": 30}),
("/openapi.json", 200, openapi_schema),
],
)
def test_get(path, expected_status, expected_response):
response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response

107
tests/test_tutorial/test_annotated/test_tutorial002_py39.py

@ -0,0 +1,107 @@
import pytest
from fastapi.testclient import TestClient
from ...utils import needs_py39
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/items/": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Items",
"operationId": "read_items_items__get",
"parameters": [
{
"required": False,
"schema": {"title": "Q", "type": "string"},
"name": "q",
"in": "query",
},
{
"required": False,
"schema": {"title": "Skip", "type": "integer", "default": 0},
"name": "skip",
"in": "query",
},
{
"required": False,
"schema": {"title": "Limit", "type": "integer", "default": 100},
"name": "limit",
"in": "query",
},
],
}
},
},
"components": {
"schemas": {
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
}
},
}
@pytest.fixture(name="client")
def get_client():
from docs_src.annotated.tutorial002_py39 import app
client = TestClient(app)
return client
@needs_py39
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[
("/items", 200, {"q": None, "skip": 0, "limit": 100}),
("/items?q=foo", 200, {"q": "foo", "skip": 0, "limit": 100}),
("/items?q=foo&skip=5", 200, {"q": "foo", "skip": 5, "limit": 100}),
("/items?q=foo&skip=5&limit=30", 200, {"q": "foo", "skip": 5, "limit": 30}),
("/openapi.json", 200, openapi_schema),
],
)
def test_get(path, expected_status, expected_response, client):
response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response

138
tests/test_tutorial/test_annotated/test_tutorial003.py

@ -0,0 +1,138 @@
import pytest
from fastapi.testclient import TestClient
from docs_src.annotated.tutorial003 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/items/{item_id}": {
"get": {
"summary": "Read Items",
"operationId": "read_items_items__item_id__get",
"parameters": [
{
"required": True,
"schema": {
"title": "Item Id",
"exclusiveMinimum": 0.0,
"type": "integer",
},
"name": "item_id",
"in": "path",
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
},
"/users": {
"get": {
"summary": "Read Users",
"operationId": "read_users_users_get",
"parameters": [
{
"required": False,
"schema": {
"title": "User Id",
"minLength": 1,
"type": "string",
"default": "me",
},
"name": "user_id",
"in": "query",
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
},
},
"components": {
"schemas": {
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
}
},
}
item_id_negative = {
"detail": [
{
"ctx": {"limit_value": 0},
"loc": ["path", "item_id"],
"msg": "ensure this value is greater than 0",
"type": "value_error.number.not_gt",
}
]
}
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[
("/items/1", 200, {"item_id": 1}),
("/items/-1", 422, item_id_negative),
("/users", 200, {"user_id": "me"}),
("/users?user_id=foo", 200, {"user_id": "foo"}),
("/openapi.json", 200, openapi_schema),
],
)
def test_get(path, expected_status, expected_response):
response = client.get(path)
assert response.status_code == expected_status, response.text
assert response.json() == expected_response

145
tests/test_tutorial/test_annotated/test_tutorial003_py39.py

@ -0,0 +1,145 @@
import pytest
from fastapi.testclient import TestClient
from ...utils import needs_py39
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/items/{item_id}": {
"get": {
"summary": "Read Items",
"operationId": "read_items_items__item_id__get",
"parameters": [
{
"required": True,
"schema": {
"title": "Item Id",
"exclusiveMinimum": 0.0,
"type": "integer",
},
"name": "item_id",
"in": "path",
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
},
"/users": {
"get": {
"summary": "Read Users",
"operationId": "read_users_users_get",
"parameters": [
{
"required": False,
"schema": {
"title": "User Id",
"minLength": 1,
"type": "string",
"default": "me",
},
"name": "user_id",
"in": "query",
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
},
},
"components": {
"schemas": {
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
}
},
}
item_id_negative = {
"detail": [
{
"ctx": {"limit_value": 0},
"loc": ["path", "item_id"],
"msg": "ensure this value is greater than 0",
"type": "value_error.number.not_gt",
}
]
}
@pytest.fixture(name="client")
def get_client():
from docs_src.annotated.tutorial003_py39 import app
client = TestClient(app)
return client
@needs_py39
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[
("/items/1", 200, {"item_id": 1}),
("/items/-1", 422, item_id_negative),
("/users", 200, {"user_id": "me"}),
("/users?user_id=foo", 200, {"user_id": "foo"}),
("/openapi.json", 200, openapi_schema),
],
)
def test_get(path, expected_status, expected_response, client):
response = client.get(path)
assert response.status_code == expected_status, response.text
assert response.json() == expected_response

0
tests/test_tutorial/test_dataclasses/__init__.py

Loading…
Cancel
Save