from copy import deepcopy
from datetime import date, datetime, time, timedelta
from decimal import Decimal
-from typing import Any, Callable, Dict, List, Mapping, Sequence, Tuple, Type
+from typing import Any, Callable, Dict, List, Mapping, Sequence, Tuple, Type, Union
from uuid import UUID
from fastapi import params
from pydantic import BaseConfig, Schema, create_model
from pydantic.error_wrappers import ErrorWrapper
from pydantic.errors import MissingError
-from pydantic.fields import Field, Required
+from pydantic.fields import Field, Required, Shape
from pydantic.schema import get_annotation_from_schema
from pydantic.utils import lenient_issubclass
from starlette.concurrency import run_in_threadpool
-from starlette.requests import Request
+from starlette.requests import Headers, QueryParams, Request
param_supported_types = (
str,
elif isinstance(param.default, params.Param):
if param.annotation != param.empty:
assert lenient_issubclass(
- param.annotation, param_supported_types
- ), f"Parameters for Path, Query, Header and Cookies must be of type str, int, float or bool: {param}"
+ param.annotation, param_supported_types + (list, tuple, set)
+ ), f"Parameters for Path, Query, Header and Cookies must be of type str, int, float, bool, list, tuple or set: {param}"
add_param_to_fields(
param=param, dependant=dependant, default_schema=params.Query
)
def request_params_to_args(
- required_params: Sequence[Field], received_params: Mapping[str, Any]
+ required_params: Sequence[Field],
+ received_params: Union[Mapping[str, Any], QueryParams, Headers],
) -> Tuple[Dict[str, Any], List[ErrorWrapper]]:
values = {}
errors = []
for field in required_params:
- value = received_params.get(field.alias)
+ if field.shape in {Shape.LIST, Shape.SET, Shape.TUPLE} and isinstance(
+ received_params, (QueryParams, Headers)
+ ):
+ value = received_params.getlist(field.alias)
+ else:
+ value = received_params.get(field.alias)
schema: params.Param = field.schema
assert isinstance(schema, params.Param), "Params must be subclasses of Param"
if value is None:
--- /dev/null
+from typing import List
+
+from fastapi import FastAPI
+from pydantic import BaseModel
+from starlette.testclient import TestClient
+
+app = FastAPI()
+
+
+class Item(BaseModel):
+ name: str
+ age: int
+
+
+@app.post("/items/")
+def save_item_no_body(item: List[Item]):
+ return {"item": item}
+
+
+client = TestClient(app)
+
+
+openapi_schema = {
+ "openapi": "3.0.2",
+ "info": {"title": "Fast API", "version": "0.1.0"},
+ "paths": {
+ "/items/": {
+ "post": {
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ "summary": "Save Item No Body Post",
+ "operationId": "save_item_no_body_items__post",
+ "requestBody": {
+ "content": {
+ "application/json": {
+ "schema": {
+ "title": "Item",
+ "type": "array",
+ "items": {"$ref": "#/components/schemas/Item"},
+ }
+ }
+ },
+ "required": True,
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "Item": {
+ "title": "Item",
+ "required": ["name", "age"],
+ "type": "object",
+ "properties": {
+ "name": {"title": "Name", "type": "string"},
+ "age": {"title": "Age", "type": "integer"},
+ },
+ },
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {"type": "string"},
+ },
+ "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"},
+ }
+ },
+ },
+ }
+ },
+}
+
+multiple_errors = {
+ "detail": [
+ {
+ "loc": ["body", "item", 0, "name"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "item", 0, "age"],
+ "msg": "value is not a valid integer",
+ "type": "type_error.integer",
+ },
+ {
+ "loc": ["body", "item", 1, "name"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "item", 1, "age"],
+ "msg": "value is not a valid integer",
+ "type": "type_error.integer",
+ },
+ ]
+}
+
+
+def test_openapi_schema():
+ response = client.get("/openapi.json")
+ assert response.status_code == 200
+ assert response.json() == openapi_schema
+
+
+def test_put_correct_body():
+ response = client.post("/items/", json=[{"name": "Foo", "age": 5}])
+ assert response.status_code == 200
+ assert response.json() == {"item": [{"name": "Foo", "age": 5}]}
+
+
+def test_put_incorrect_body():
+ response = client.post("/items/", json=[{"age": "five"}, {"age": "six"}])
+ assert response.status_code == 422
+ assert response.json() == multiple_errors
--- /dev/null
+from typing import List
+
+from fastapi import FastAPI, Query
+from starlette.testclient import TestClient
+
+app = FastAPI()
+
+
+@app.get("/items/")
+def read_items(q: List[int] = Query(None)):
+ return {"q": q}
+
+
+client = TestClient(app)
+
+
+openapi_schema = {
+ "openapi": "3.0.2",
+ "info": {"title": "Fast API", "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 Get",
+ "operationId": "read_items_items__get",
+ "parameters": [
+ {
+ "required": False,
+ "schema": {
+ "title": "Q",
+ "type": "array",
+ "items": {"type": "integer"},
+ },
+ "name": "q",
+ "in": "query",
+ }
+ ],
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {"type": "string"},
+ },
+ "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"},
+ }
+ },
+ },
+ }
+ },
+}
+
+multiple_errors = {
+ "detail": [
+ {
+ "loc": ["query", "q", 0],
+ "msg": "value is not a valid integer",
+ "type": "type_error.integer",
+ },
+ {
+ "loc": ["query", "q", 1],
+ "msg": "value is not a valid integer",
+ "type": "type_error.integer",
+ },
+ ]
+}
+
+
+def test_openapi_schema():
+ response = client.get("/openapi.json")
+ assert response.status_code == 200
+ assert response.json() == openapi_schema
+
+
+def test_multi_query():
+ response = client.get("/items/?q=5&q=6")
+ assert response.status_code == 200
+ assert response.json() == {"q": [5, 6]}
+
+
+def test_multi_query_incorrect():
+ response = client.get("/items/?q=five&q=six")
+ assert response.status_code == 422
+ assert response.json() == multiple_errors
--- /dev/null
+from fastapi import FastAPI
+from starlette.testclient import TestClient
+
+app = FastAPI()
+
+
+@app.put("/items/{item_id}")
+def save_item_no_body(item_id: str):
+ return {"item_id": item_id}
+
+
+client = TestClient(app)
+
+
+openapi_schema = {
+ "openapi": "3.0.2",
+ "info": {"title": "Fast API", "version": "0.1.0"},
+ "paths": {
+ "/items/{item_id}": {
+ "put": {
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ "summary": "Save Item No Body Put",
+ "operationId": "save_item_no_body_items__item_id__put",
+ "parameters": [
+ {
+ "required": True,
+ "schema": {"title": "Item_Id", "type": "string"},
+ "name": "item_id",
+ "in": "path",
+ }
+ ],
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {"type": "string"},
+ },
+ "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"},
+ }
+ },
+ },
+ }
+ },
+}
+
+
+def test_openapi_schema():
+ response = client.get("/openapi.json")
+ assert response.status_code == 200
+ assert response.json() == openapi_schema
+
+
+def test_put_no_body():
+ response = client.put("/items/foo")
+ assert response.status_code == 200
+ assert response.json() == {"item_id": "foo"}
+
+
+def test_put_no_body_with_body():
+ response = client.put("/items/foo", json={"name": "Foo"})
+ assert response.status_code == 200
+ assert response.json() == {"item_id": "foo"}
projects / thirdparty / fastapi / fastapi.git / commitdiff
