--- /dev/null
+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
--- /dev/null
+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
--- /dev/null
+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
--- /dev/null
+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
--- /dev/null
+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}
--- /dev/null
+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}
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
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,
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,
)
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):
+ if add_non_field_param_to_dependency(
+ param_name=param_name,
+ type_annotation=type_annotation,
+ dependant=dependant,
+ ):
+ assert (
+ param_field is None
+ ), f"Cannot specify multiple FastAPI annotations for {param_name!r}"
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,
- 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()"
+ 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
- if (
+ 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)
- if not field_info.alias and getattr(field_info, "convert_underscores", None):
- alias = param.name.replace("_", "-")
- else:
- alias = field_info.alias or param.name
- field = create_response_field(
- name=param.name,
- type_=annotation,
- default=default_value,
- alias=alias,
- required=required,
- 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):
- field.field_info = params.File(field_info.default)
+ 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("_", "-")
+ else:
+ alias = field_info.alias or param_name
+ field = create_response_field(
+ name=param_name,
+ type_=annotation,
+ default=field_info.default,
+ alias=alias,
+ required=field_info.default in (Required, Undefined),
+ field_info=field_info,
+ )
+ 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:
def Path( # noqa: N802
- default: Any = Undefined,
+ default: Any = ...,
*,
alias: Optional[str] = None,
title: Optional[str] = None,
def __init__(
self,
- default: Any = Undefined,
+ default: Any = ...,
*,
alias: Optional[str] = None,
title: Optional[str] = None,
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,
class Form(Body):
def __init__(
self,
- default: Any,
+ default: Any = Undefined,
*,
media_type: str = "application/x-www-form-urlencoded",
alias: Optional[str] = None,
class File(Form):
def __init__(
self,
- default: Any,
+ default: Any = Undefined,
*,
media_type: str = "multipart/form-data",
alias: Optional[str] = None,
-import functools
import re
import warnings
from dataclasses import is_dataclass
class_validators = class_validators or {}
field_info = field_info or FieldInfo()
- response_field = functools.partial(
- ModelField,
- name=name,
- type_=type_,
- class_validators=class_validators,
- default=default,
- required=required,
- model_config=model_config,
- alias=alias,
- )
-
try:
- return response_field(field_info=field_info)
+ return ModelField(
+ name=name,
+ type_=type_,
+ class_validators=class_validators,
+ default=default,
+ required=required,
+ model_config=model_config,
+ alias=alias,
+ field_info=field_info,
+ )
except RuntimeError:
raise fastapi.exceptions.FastAPIError(
"Invalid args for response field! Hint: "
@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
--- /dev/null
+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
--- /dev/null
+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
],
}
},
- "/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": {
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):
("/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"),
--- /dev/null
+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
--- /dev/null
+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
--- /dev/null
+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
--- /dev/null
+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
--- /dev/null
+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
--- /dev/null
+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
projects / thirdparty / fastapi / fastapi.git / commitdiff
