from copy import deepcopy
from typing import Any, Callable, Dict, List, Mapping, Sequence, Tuple, Type
-from fastapi import params
-from fastapi.dependencies.models import Dependant, SecurityRequirement
-from fastapi.security.base import SecurityBase
-from fastapi.utils import get_path_param_names
from pydantic import BaseConfig, Schema, create_model
from pydantic.error_wrappers import ErrorWrapper
from pydantic.errors import MissingError
from starlette.concurrency import run_in_threadpool
from starlette.requests import Request
+from fastapi import params
+from fastapi.dependencies.models import Dependant, SecurityRequirement
+from fastapi.security.base import SecurityBase
+from fastapi.utils import get_path_param_names
+
param_supported_types = (str, int, float, bool)
embed = getattr(field.schema, "embed", None)
if len(required_params) == 1 and not embed:
received_body = {field.alias: received_body}
+ elif received_body is None:
+ received_body = {}
for field in required_params:
value = received_body.get(field.alias)
if value is None:
import logging
from typing import Any, Callable, List, Optional, Type
-from fastapi import params
-from fastapi.dependencies.models import Dependant
-from fastapi.dependencies.utils import get_body_field, get_dependant, solve_dependencies
-from fastapi.encoders import jsonable_encoder
from pydantic import BaseConfig, BaseModel, Schema
from pydantic.error_wrappers import ErrorWrapper, ValidationError
from pydantic.fields import Field
from starlette.routing import get_name, request_response
from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY
+from fastapi import params
+from fastapi.dependencies.models import Dependant
+from fastapi.dependencies.utils import get_body_field, get_dependant, solve_dependencies
+from fastapi.encoders import jsonable_encoder
+
def serialize_response(*, field: Field = None, response: Response) -> Any:
if field:
try:
body = None
if body_field:
- if is_body_form:
+ body_bytes = await request.body()
+ if body_bytes and is_body_form:
raw_body = await request.form()
body = {}
for field, value in raw_body.items():
body[field] = await value.read()
else:
body[field] = value
- else:
+ elif body_bytes:
body = await request.json()
except Exception as e:
logging.error("Error getting request body", e)