)
from fastapi.openapi.utils import get_openapi
from fastapi.params import Depends
-from fastapi.utils import warning_response_model_skip_defaults_deprecated
from starlette.applications import Starlette
from starlette.datastructures import State
from starlette.exceptions import HTTPException
methods: Optional[List[str]] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
response_class: Optional[Type[Response]] = None,
name: Optional[str] = None,
) -> None:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
self.router.add_api_route(
path,
endpoint=endpoint,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
methods: Optional[List[str]] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
response_class: Optional[Type[Response]] = None,
name: Optional[str] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
-
def decorator(func: Callable) -> Callable:
self.router.add_api_route(
path,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[routing.APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.router.get(
path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[routing.APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.router.put(
path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[routing.APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.router.post(
path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[routing.APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.router.delete(
path,
response_model=response_model,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
operation_id=operation_id,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[routing.APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.router.options(
path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[routing.APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.router.head(
path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[routing.APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.router.patch(
path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[routing.APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.router.trace(
path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
from typing import Callable, List, Optional, Sequence
from fastapi.security.base import SecurityBase
-
-try:
- from pydantic.fields import ModelField
-except ImportError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- from pydantic.fields import Field as ModelField # type: ignore
+from pydantic.fields import ModelField
param_supported_types = (str, int, float, bool)
from fastapi.security.base import SecurityBase
from fastapi.security.oauth2 import OAuth2, SecurityScopes
from fastapi.security.open_id_connect_url import OpenIdConnect
-from fastapi.utils import (
- PYDANTIC_1,
- create_response_field,
- get_field_info,
- get_path_param_names,
-)
-from pydantic import BaseConfig, BaseModel, create_model
+from fastapi.utils import create_response_field, get_path_param_names
+from pydantic import BaseModel, create_model
from pydantic.error_wrappers import ErrorWrapper
from pydantic.errors import MissingError
+from pydantic.fields import (
+ SHAPE_LIST,
+ SHAPE_SEQUENCE,
+ SHAPE_SET,
+ SHAPE_SINGLETON,
+ SHAPE_TUPLE,
+ SHAPE_TUPLE_ELLIPSIS,
+ FieldInfo,
+ ModelField,
+ Required,
+)
+from pydantic.schema import get_annotation_from_field_info
+from pydantic.typing import ForwardRef, evaluate_forwardref
from pydantic.utils import lenient_issubclass
from starlette.background import BackgroundTasks
from starlette.concurrency import run_in_threadpool
from starlette.responses import Response
from starlette.websockets import WebSocket
-try:
- from pydantic.fields import (
- SHAPE_LIST,
- SHAPE_SEQUENCE,
- SHAPE_SET,
- SHAPE_SINGLETON,
- SHAPE_TUPLE,
- SHAPE_TUPLE_ELLIPSIS,
- FieldInfo,
- ModelField,
- Required,
- )
- from pydantic.schema import get_annotation_from_field_info
- from pydantic.typing import ForwardRef, evaluate_forwardref
-except ImportError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- from pydantic import Schema as FieldInfo # type: ignore
- from pydantic.fields import Field as ModelField # type: ignore
- from pydantic.fields import Required, Shape # type: ignore
- from pydantic.schema import get_annotation_from_schema # type: ignore
- from pydantic.utils import ForwardRef, evaluate_forwardref # type: ignore
-
- SHAPE_LIST = Shape.LIST
- SHAPE_SEQUENCE = Shape.SEQUENCE
- SHAPE_SET = Shape.SET
- SHAPE_SINGLETON = Shape.SINGLETON
- SHAPE_TUPLE = Shape.TUPLE
- SHAPE_TUPLE_ELLIPSIS = Shape.TUPLE_ELLIPS
-
- def get_annotation_from_field_info(
- annotation: Any, field_info: FieldInfo, field_name: str
- ) -> Type[Any]:
- return get_annotation_from_schema(annotation, field_info)
-
-
sequence_shapes = {
SHAPE_LIST,
SHAPE_SET,
def check_file_field(field: ModelField) -> None:
- field_info = get_field_info(field)
+ field_info = field.field_info
if isinstance(field_info, params.Form):
try:
# __version__ is available in both multiparts, and can be mocked
def is_scalar_field(field: ModelField) -> bool:
- field_info = get_field_info(field)
+ field_info = field.field_info
if not (
field.shape == SHAPE_SINGLETON
and not lenient_issubclass(field.type_, BaseModel)
) and is_scalar_sequence_field(param_field):
add_param_to_fields(field=param_field, dependant=dependant)
else:
- field_info = get_field_info(param_field)
+ 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(...)"
)
field.required = required
if not had_schema and not is_scalar_field(field=field):
- if PYDANTIC_1:
- field.field_info = params.Body(field_info.default)
- else:
- field.schema = params.Body(field_info.default) # type: ignore # pragma: nocover
+ field.field_info = params.Body(field_info.default)
return field
def add_param_to_fields(*, field: ModelField, dependant: Dependant) -> None:
- field_info = cast(params.Param, get_field_info(field))
+ field_info = cast(params.Param, field.field_info)
if field_info.in_ == params.ParamTypes.path:
dependant.path_params.append(field)
elif field_info.in_ == params.ParamTypes.query:
value = received_params.getlist(field.alias) or field.default
else:
value = received_params.get(field.alias)
- field_info = get_field_info(field)
+ field_info = field.field_info
assert isinstance(
field_info, params.Param
), "Params must be subclasses of Param"
if value is None:
if field.required:
- if PYDANTIC_1:
- errors.append(
- ErrorWrapper(
- MissingError(), loc=(field_info.in_.value, field.alias)
- )
- )
- else: # pragma: nocover
- errors.append(
- ErrorWrapper( # type: ignore
- MissingError(),
- loc=(field_info.in_.value, field.alias),
- config=BaseConfig,
- )
+ errors.append(
+ ErrorWrapper(
+ MissingError(), loc=(field_info.in_.value, field.alias)
)
+ )
else:
values[field.name] = deepcopy(field.default)
continue
errors = []
if required_params:
field = required_params[0]
- field_info = get_field_info(field)
+ field_info = field.field_info
embed = getattr(field_info, "embed", None)
field_alias_omitted = len(required_params) == 1 and not embed
if field_alias_omitted:
def get_missing_field_error(loc: Tuple[str, ...]) -> ErrorWrapper:
- if PYDANTIC_1:
- missing_field_error = ErrorWrapper(MissingError(), loc=loc)
- else: # pragma: no cover
- missing_field_error = ErrorWrapper( # type: ignore
- MissingError(), loc=loc, config=BaseConfig,
- )
+ missing_field_error = ErrorWrapper(MissingError(), loc=loc)
return missing_field_error
default=field.default,
required=field.required,
alias=field.alias,
- field_info=get_field_info(field),
+ field_info=field.field_info,
)
return out_field
if not flat_dependant.body_params:
return None
first_param = flat_dependant.body_params[0]
- field_info = get_field_info(first_param)
+ field_info = first_param.field_info
embed = getattr(field_info, "embed", None)
body_param_names_set = {param.name for param in flat_dependant.body_params}
if len(body_param_names_set) == 1 and not embed:
# in case a sub-dependency is evaluated with a single unique body field
# That is combined (embedded) with other body fields
for param in flat_dependant.body_params:
- setattr(get_field_info(param), "embed", True)
+ setattr(param.field_info, "embed", True)
model_name = "Body_" + name
BodyModel = create_model(model_name)
for f in flat_dependant.body_params:
required = any(True for f in flat_dependant.body_params if f.required)
BodyFieldInfo_kwargs: Dict[str, Any] = dict(default=None)
- if any(
- isinstance(get_field_info(f), params.File) for f in flat_dependant.body_params
- ):
+ if any(isinstance(f.field_info, params.File) for f in flat_dependant.body_params):
BodyFieldInfo: Type[params.Body] = params.File
- elif any(
- isinstance(get_field_info(f), params.Form) for f in flat_dependant.body_params
- ):
+ elif any(isinstance(f.field_info, params.Form) for f in flat_dependant.body_params):
BodyFieldInfo = params.Form
else:
BodyFieldInfo = params.Body
body_param_media_types = [
- getattr(get_field_info(f), "media_type")
+ getattr(f.field_info, "media_type")
for f in flat_dependant.body_params
- if isinstance(get_field_info(f), params.Body)
+ if isinstance(f.field_info, params.Body)
]
if len(set(body_param_media_types)) == 1:
BodyFieldInfo_kwargs["media_type"] = body_param_media_types[0]
from types import GeneratorType
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
-from fastapi.logger import logger
-from fastapi.utils import PYDANTIC_1
from pydantic import BaseModel
from pydantic.json import ENCODERS_BY_TYPE
def jsonable_encoder(
obj: Any,
include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
by_alias: bool = True,
- skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
custom_encoder: dict = {},
sqlalchemy_safe: bool = True,
) -> Any:
- if skip_defaults is not None:
- logger.warning( # pragma: nocover
- "skip_defaults in jsonable_encoder has been deprecated in favor of "
- "exclude_unset to keep in line with Pydantic v1, support for it will be "
- "removed soon."
- )
if include is not None and not isinstance(include, set):
include = set(include)
if exclude is not None and not isinstance(exclude, set):
encoder = getattr(obj.__config__, "json_encoders", {})
if custom_encoder:
encoder.update(custom_encoder)
- if PYDANTIC_1:
- obj_dict = obj.dict(
- include=include,
- exclude=exclude,
- by_alias=by_alias,
- exclude_unset=bool(exclude_unset or skip_defaults),
- exclude_none=exclude_none,
- exclude_defaults=exclude_defaults,
- )
- else: # pragma: nocover
- if exclude_defaults:
- raise ValueError("Cannot use exclude_defaults")
- obj_dict = obj.dict(
- include=include,
- exclude=exclude,
- by_alias=by_alias,
- skip_defaults=bool(exclude_unset or skip_defaults),
- )
+ obj_dict = obj.dict(
+ include=include,
+ exclude=exclude,
+ by_alias=by_alias,
+ exclude_unset=exclude_unset,
+ exclude_none=exclude_none,
+ exclude_defaults=exclude_defaults,
+ )
if "__root__" in obj_dict:
obj_dict = obj_dict["__root__"]
return jsonable_encoder(
or (not key.startswith("_sa"))
)
and (value is not None or not exclude_none)
- and ((include and key in include) or key not in exclude)
+ and ((include and key in include) or not exclude or key not in exclude)
):
encoded_key = jsonable_encoder(
key,
from typing import Any, Dict, Optional, Sequence
-from fastapi.utils import PYDANTIC_1
from pydantic import ValidationError, create_model
from pydantic.error_wrappers import ErrorList
from starlette.exceptions import HTTPException as StarletteHTTPException
-from starlette.requests import Request
-from starlette.websockets import WebSocket
class HTTPException(StarletteHTTPException):
class RequestValidationError(ValidationError):
def __init__(self, errors: Sequence[ErrorList], *, body: Any = None) -> None:
self.body = body
- if PYDANTIC_1:
- super().__init__(errors, RequestErrorModel)
- else:
- super().__init__(errors, Request) # type: ignore # pragma: nocover
+ super().__init__(errors, RequestErrorModel)
class WebSocketRequestValidationError(ValidationError):
def __init__(self, errors: Sequence[ErrorList]) -> None:
- if PYDANTIC_1:
- super().__init__(errors, WebSocketErrorModel)
- else:
- super().__init__(errors, WebSocket) # type: ignore # pragma: nocover
+ super().__init__(errors, WebSocketErrorModel)
from typing import Any, Callable, Dict, Iterable, List, Optional, Union
from fastapi.logger import logger
-from pydantic import BaseModel
-
-try:
- from pydantic import AnyUrl, Field
-except ImportError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- from pydantic import Schema as Field # type: ignore
- from pydantic import UrlStr as AnyUrl # type: ignore
+from pydantic import AnyUrl, BaseModel, Field
try:
import email_validator
assert email_validator # make autoflake ignore the unused import
- try:
- from pydantic import EmailStr
- except ImportError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- from pydantic.types import EmailStr # type: ignore
+ from pydantic import EmailStr
except ImportError: # pragma: no cover
class EmailStr(str): # type: ignore
from fastapi.utils import (
deep_dict_update,
generate_operation_id_for_path,
- get_field_info,
get_model_definitions,
)
from pydantic import BaseModel
+from pydantic.fields import ModelField
from pydantic.schema import (
field_schema,
get_flat_models_from_fields,
from starlette.routing import BaseRoute
from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY
-try:
- from pydantic.fields import ModelField
-except ImportError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- from pydantic.fields import Field as ModelField # type: ignore
-
validation_error_definition = {
"title": "ValidationError",
"type": "object",
) -> List[Dict[str, Any]]:
parameters = []
for param in all_route_params:
- field_info = get_field_info(param)
+ field_info = param.field_info
field_info = cast(Param, field_info)
# ignore mypy error until enum schemas are released
parameter = {
body_schema, _, _ = field_schema(
body_field, model_name_map=model_name_map, ref_prefix=REF_PREFIX # type: ignore
)
- field_info = cast(Body, get_field_info(body_field))
+ field_info = cast(Body, body_field.field_info)
request_media_type = field_info.media_type
required = body_field.required
request_body_oai: Dict[str, Any] = {}
from enum import Enum
from typing import Any, Callable, Optional, Sequence
-try:
- from pydantic.fields import FieldInfo
-except ImportError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- from pydantic import Schema as FieldInfo # type: ignore
+from pydantic.fields import FieldInfo
class ParamTypes(Enum):
-from starlette.requests import HTTPConnection, Request # noqa
+from starlette.requests import HTTPConnection as HTTPConnection # noqa: F401
+from starlette.requests import Request as Request # noqa: F401
from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError
from fastapi.openapi.constants import STATUS_CODES_WITH_NO_BODY
from fastapi.utils import (
- PYDANTIC_1,
create_cloned_field,
create_response_field,
generate_operation_id_for_path,
- get_field_info,
- warning_response_model_skip_defaults_deprecated,
)
from pydantic import BaseModel
from pydantic.error_wrappers import ErrorWrapper, ValidationError
+from pydantic.fields import ModelField
from starlette import routing
from starlette.concurrency import run_in_threadpool
from starlette.exceptions import HTTPException
from starlette.types import ASGIApp
from starlette.websockets import WebSocket
-try:
- from pydantic.fields import ModelField
-except ImportError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- from pydantic.fields import Field as ModelField # type: ignore
-
def _prepare_response_content(
res: Any,
exclude_none: bool = False,
) -> Any:
if isinstance(res, BaseModel):
- if PYDANTIC_1:
- return res.dict(
- by_alias=True,
- exclude_unset=exclude_unset,
- exclude_defaults=exclude_defaults,
- exclude_none=exclude_none,
- )
- else:
- return res.dict(
- by_alias=True, skip_defaults=exclude_unset,
- ) # pragma: nocover
+ return res.dict(
+ by_alias=True,
+ exclude_unset=exclude_unset,
+ exclude_defaults=exclude_defaults,
+ exclude_none=exclude_none,
+ )
elif isinstance(res, list):
return [
_prepare_response_content(
field: Optional[ModelField] = None,
response_content: Any,
include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
by_alias: bool = True,
exclude_unset: bool = False,
exclude_defaults: bool = False,
response_class: Type[Response] = JSONResponse,
response_field: Optional[ModelField] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
) -> Callable:
assert dependant.call is not None, "dependant.call must be a function"
is_coroutine = asyncio.iscoroutinefunction(dependant.call)
- is_body_form = body_field and isinstance(get_field_info(body_field), params.Form)
+ is_body_form = body_field and isinstance(body_field.field_info, params.Form)
async def app(request: Request) -> Response:
try:
methods: Optional[Union[Set[str], List[str]]] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
methods: Optional[Union[Set[str], List[str]]] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
route_class_override: Optional[Type[APIRoute]] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> None:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
route_class = route_class_override or self.route_class
route = route_class(
path,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
methods: Optional[List[str]] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
-
def decorator(func: Callable) -> Callable:
self.add_api_route(
path,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.api_route(
path=path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.api_route(
path=path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.api_route(
path=path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.api_route(
path=path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.api_route(
path=path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.api_route(
path=path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
return self.api_route(
path=path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
deprecated: Optional[bool] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
- response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
+ response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
- response_model_skip_defaults: Optional[bool] = None,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
name: Optional[str] = None,
callbacks: Optional[List[APIRoute]] = None,
) -> Callable:
- if response_model_skip_defaults is not None:
- warning_response_model_skip_defaults_deprecated() # pragma: nocover
+
return self.api_route(
path=path,
response_model=response_model,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
- response_model_exclude_unset=bool(
- response_model_exclude_unset or response_model_skip_defaults
- ),
+ response_model_exclude_unset=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
from typing import Any, Dict, Optional, Set, Type, Union, cast
import fastapi
-from fastapi.logger import logger
from fastapi.openapi.constants import REF_PREFIX
from pydantic import BaseConfig, BaseModel, create_model
from pydantic.class_validators import Validator
+from pydantic.fields import FieldInfo, ModelField, UndefinedType
from pydantic.schema import model_process_schema
from pydantic.utils import lenient_issubclass
-try:
- from pydantic.fields import FieldInfo, ModelField, UndefinedType
-
- PYDANTIC_1 = True
-except ImportError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- from pydantic import Schema as FieldInfo # type: ignore
- from pydantic.fields import Field as ModelField # type: ignore
-
- class UndefinedType: # type: ignore
- def __repr__(self) -> str:
- return "PydanticUndefined"
-
- logger.warning(
- "Pydantic versions < 1.0.0 are deprecated in FastAPI and support will be "
- "removed soon."
- )
- PYDANTIC_1 = False
-
-
-# TODO: remove when removing support for Pydantic < 1.0.0
-def get_field_info(field: ModelField) -> FieldInfo:
- if PYDANTIC_1:
- return field.field_info # type: ignore
- else:
- return field.schema # type: ignore # pragma: nocover
-
-
-# TODO: remove when removing support for Pydantic < 1.0.0
-def warning_response_model_skip_defaults_deprecated() -> None:
- logger.warning( # pragma: nocover
- "response_model_skip_defaults has been deprecated in favor of "
- "response_model_exclude_unset to keep in line with Pydantic v1, support for "
- "it will be removed soon."
- )
-
def get_model_definitions(
*,
)
try:
- if PYDANTIC_1:
- return response_field(field_info=field_info)
- else: # pragma: nocover
- return response_field(schema=field_info)
+ return response_field(field_info=field_info)
except RuntimeError:
raise fastapi.exceptions.FastAPIError(
f"Invalid args for response field! Hint: check that {type_} is a valid pydantic field type"
new_field.default = field.default
new_field.required = field.required
new_field.model_config = field.model_config
- if PYDANTIC_1:
- new_field.field_info = field.field_info
- else: # pragma: nocover
- new_field.schema = field.schema # type: ignore
+ new_field.field_info = field.field_info
new_field.allow_none = field.allow_none
new_field.validate_always = field.validate_always
if field.sub_fields:
field.key_field, cloned_types=cloned_types
)
new_field.validators = field.validators
- if PYDANTIC_1:
- new_field.pre_validators = field.pre_validators
- new_field.post_validators = field.post_validators
- else: # pragma: nocover
- new_field.whole_pre_validators = field.whole_pre_validators # type: ignore
- new_field.whole_post_validators = field.whole_post_validators # type: ignore
+ new_field.pre_validators = field.pre_validators
+ new_field.post_validators = field.post_validators
new_field.parse_json = field.parse_json
new_field.shape = field.shape
- try:
- new_field.populate_validators()
- except AttributeError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- new_field._populate_validators() # type: ignore
+ new_field.populate_validators()
return new_field
]
requires = [
"starlette ==0.13.6",
- "pydantic >=0.32.2,<2.0.0"
+ "pydantic >=1.0.0,<2.0.0"
]
description-file = "README.md"
requires-python = ">=3.6"
import pytest
from fastapi.encoders import jsonable_encoder
-from pydantic import BaseModel, ValidationError, create_model
-
-try:
- from pydantic import Field
-except ImportError: # pragma: nocover
- # TODO: remove when removing support for Pydantic < 1.0.0
- from pydantic import Schema as Field
+from pydantic import BaseModel, Field, ValidationError, create_model
class Person:
from docs_src.body_fields.tutorial001 import app
-# TODO: remove when removing support for Pydantic < 1.0.0
-try:
- from pydantic import Field # noqa
-except ImportError: # pragma: nocover
- import pydantic
-
- pydantic.Field = pydantic.Schema
-
-
client = TestClient(app)
projects / thirdparty / fastapi / fastapi.git / commitdiff
