--- /dev/null
+from datetime import datetime, time, timedelta
+from uuid import UUID
+
+from fastapi import Body, FastAPI
+
+app = FastAPI()
+
+
+@app.put("/items/{item_id}")
+async def read_items(
+ item_id: UUID,
+ start_datetime: datetime = Body(None),
+ end_datetime: datetime = Body(None),
+ repeat_at: time = Body(None),
+ process_after: timedelta = Body(None),
+):
+ start_process = start_datetime + process_after
+ duration = end_datetime - start_process
+ return {
+ "item_id": item_id,
+ "start_datetime": start_datetime,
+ "end_datetime": end_datetime,
+ "repeat_at": repeat_at,
+ "process_after": process_after,
+ "start_process": start_process,
+ "duration": duration,
+ }
Apart from normal singular types like `str`, `int`, `float`, etc. You can use more complex singular types that inherit from `str`.
-To see all the options you have, checkout the docs for <a href="https://pydantic-docs.helpmanual.io/#exotic-types" target="_blank">Pydantic's exotic types</a>.
+To see all the options you have, checkout the docs for <a href="https://pydantic-docs.helpmanual.io/#exotic-types" target="_blank">Pydantic's exotic types</a>. You will see some examples in the next chapter.
For example, as in the `Image` model we have a `url` field, we can declare it to be instead of a `str`, a Pydantic's `UrlStr`:
--- /dev/null
+Up to now, you have been using common data types, like:
+
+* `int`
+* `float`
+* `str`
+* `bool`
+
+But you can also use more complex data types.
+
+And you will still have the same features as seen up to now:
+
+* Great editor support.
+* Data conversion from incoming requests.
+* Data conversion for response data.
+* Data validation.
+* Automatic annotation and documentation.
+
+## Other data types
+
+Here are some of the additional data types you can use:
+
+* `UUID`:
+ * A standard "Universally Unique Identifier", common as an ID in many databases and systems.
+ * In requests and responses will be represented as a `str`.
+* `datetime.datetime`:
+ * A Python `datetime.datetime`.
+ * In requests and responses will be represented as a `str` in ISO 8601 format, like: `2008-09-15T15:53:00+05:00`.
+* `datetime.date`:
+ * Python `datetime.date`.
+ * In requests and responses will be represented as a `str` in ISO 8601 format, like: `2008-09-15`.
+* `datetime.time`:
+ * A Python `datetime.time`.
+ * In requests and responses will be represented as a `str` in ISO 8601 format, like: `14:23:55.003`.
+* `datetime.timedelta`:
+ * A Python `datetime.timedelta`.
+ * In requests and responses will be represented as a `float` of total seconds.
+ * Pydantic also allows representing it as a "ISO 8601 time diff encoding", <a href="https://pydantic-docs.helpmanual.io/#json-serialisation" target="_blank">see the docs for more info</a>.
+* `frozenset`:
+ * In requests and responses, treated the same as a `set`:
+ * In requests, a list will be read, eliminating duplicates and converting it to a `set`.
+ * In responses, the `set` will be converted to a `list`.
+ * The generated schema will specify that the `set` values are unique (using JSON Schema's `uniqueItems`).
+* `bytes`:
+ * Standard Python `bytes`.
+ * In requests and responses will be treated as `str`.
+ * The generated schema will specify that it's a `str` with `binary` "format".
+* `Decimal`:
+ * Standard Python `Decimal`.
+ * In requests and responses, handled the same as a `float`.
+
+
+## Example
+
+Here's an example path operation with parameters using some of the above types.
+
+```Python hl_lines="1 2 11 12 13 14 15"
+{!./src/extra_data_types/tutorial001.py!}
+```
+
+Note that the parameters inside the function have their natural data type, and you can, for example, perform normal date manipulations, like:
+
+```Python hl_lines="17 18"
+{!./src/extra_data_types/tutorial001.py!}
+```
!!! check
This will give you editor support inside of your function, with error checks, completion, etc.
-## Data "parsing"
+## Data <abbr title="also known as: serialization, parsing, marshalling">conversion</abbr>
If you run this example and open your browser at <a href="http://127.0.0.1:8000/items/3" target="_blank">http://127.0.0.1:8000/items/3</a>, you will see a response of:
import asyncio
import inspect
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 uuid import UUID
from fastapi import params
from fastapi.dependencies.models import Dependant, SecurityRequirement
from starlette.concurrency import run_in_threadpool
from starlette.requests import Request
-param_supported_types = (str, int, float, bool)
+param_supported_types = (
+ str,
+ int,
+ float,
+ bool,
+ UUID,
+ date,
+ datetime,
+ time,
+ timedelta,
+ Decimal,
+)
def get_sub_dependant(*, param: inspect.Parameter, path: str) -> Dependant:
assert (
lenient_issubclass(param.annotation, param_supported_types)
or param.annotation == param.empty
- ), f"Path params must be of type str, int, float or boot: {param}"
+ ), f"Path params must be of one of the supported types"
param = signature_params[param_name]
add_param_to_fields(
param=param,
exclude: Set[str] = set(),
by_alias: bool = False,
include_none: bool = True,
+ root_encoder: bool = True,
+) -> Any:
+ errors = []
+ try:
+ return known_data_encoder(
+ obj,
+ include=include,
+ exclude=exclude,
+ by_alias=by_alias,
+ include_none=include_none,
+ )
+ except Exception as e:
+ if not root_encoder:
+ raise e
+ errors.append(e)
+ try:
+ data = dict(obj)
+ return jsonable_encoder(
+ data,
+ include=include,
+ exclude=exclude,
+ by_alias=by_alias,
+ include_none=include_none,
+ root_encoder=False,
+ )
+ except Exception as e:
+ if not root_encoder:
+ raise e
+ errors.append(e)
+ try:
+ data = vars(obj)
+ return jsonable_encoder(
+ data,
+ include=include,
+ exclude=exclude,
+ by_alias=by_alias,
+ include_none=include_none,
+ root_encoder=False,
+ )
+ except Exception as e:
+ if not root_encoder:
+ raise e
+ errors.append(e)
+ raise ValueError(errors)
+
+
+def known_data_encoder(
+ obj: Any,
+ include: Set[str] = None,
+ exclude: Set[str] = set(),
+ by_alias: bool = False,
+ include_none: bool = True,
) -> Any:
if isinstance(obj, BaseModel):
return jsonable_encoder(
obj.dict(include=include, exclude=exclude, by_alias=by_alias),
include_none=include_none,
+ root_encoder=False,
)
if isinstance(obj, Enum):
return obj.value
if isinstance(obj, dict):
return {
jsonable_encoder(
- key, by_alias=by_alias, include_none=include_none
- ): jsonable_encoder(value, by_alias=by_alias, include_none=include_none)
+ key, by_alias=by_alias, include_none=include_none, root_encoder=False
+ ): jsonable_encoder(
+ value, by_alias=by_alias, include_none=include_none, root_encoder=False
+ )
for key, value in obj.items()
if value is not None or include_none
}
exclude=exclude,
by_alias=by_alias,
include_none=include_none,
+ root_encoder=False,
)
for item in obj
]
def serialize_response(*, field: Field = None, response: Response) -> Any:
+ encoded = jsonable_encoder(response)
if field:
errors = []
- value, errors_ = field.validate(response, {}, loc=("response",))
+ value, errors_ = field.validate(encoded, {}, loc=("response",))
if isinstance(errors_, ErrorWrapper):
errors.append(errors_)
elif isinstance(errors_, list):
raise ValidationError(errors)
return jsonable_encoder(value)
else:
- return jsonable_encoder(response)
+ return encoded
def get_app(
raw_response = await run_in_threadpool(dependant.call, **values)
if isinstance(raw_response, Response):
return raw_response
- if isinstance(raw_response, BaseModel):
- return content_type(
- content=serialize_response(
- field=response_field, response=raw_response
- ),
- status_code=status_code,
- )
- errors = []
- try:
- return content_type(
- content=serialize_response(
- field=response_field, response=raw_response
- ),
- status_code=status_code,
- )
- except Exception as e:
- errors.append(e)
- try:
- response = dict(raw_response)
- return content_type(
- content=serialize_response(field=response_field, response=response),
- status_code=status_code,
- )
- except Exception as e:
- errors.append(e)
- try:
- response = vars(raw_response)
- return content_type(
- content=serialize_response(field=response_field, response=response),
- status_code=status_code,
- )
- except Exception as e:
- errors.append(e)
- raise ValueError(errors)
+ response_data = serialize_response(
+ field=response_field, response=raw_response
+ )
+ return content_type(content=response_data, status_code=status_code)
return app
- Body - Multiple Parameters: 'tutorial/body-multiple-params.md'
- Body - Schema: 'tutorial/body-schema.md'
- Body - Nested Models: 'tutorial/body-nested-models.md'
+ - Extra data types: 'tutorial/extra-data-types.md'
- Cookie Parameters: 'tutorial/cookie-params.md'
- Header Parameters: 'tutorial/header-params.md'
- Response Model: 'tutorial/response-model.md'
--- /dev/null
+from starlette.testclient import TestClient
+
+from extra_data_types.tutorial001 import app
+
+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": "Read Items Put",
+ "operationId": "read_items_items__item_id__put",
+ "parameters": [
+ {
+ "required": True,
+ "schema": {
+ "title": "Item_Id",
+ "type": "string",
+ "format": "uuid",
+ },
+ "name": "item_id",
+ "in": "path",
+ }
+ ],
+ "requestBody": {
+ "content": {
+ "application/json": {
+ "schema": {"$ref": "#/components/schemas/Body_read_items"}
+ }
+ }
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "Body_read_items": {
+ "title": "Body_read_items",
+ "type": "object",
+ "properties": {
+ "start_datetime": {
+ "title": "Start_Datetime",
+ "type": "string",
+ "format": "date-time",
+ },
+ "end_datetime": {
+ "title": "End_Datetime",
+ "type": "string",
+ "format": "date-time",
+ },
+ "repeat_at": {
+ "title": "Repeat_At",
+ "type": "string",
+ "format": "time",
+ },
+ "process_after": {
+ "title": "Process_After",
+ "type": "string",
+ "format": "time-delta",
+ },
+ },
+ },
+ "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_extra_types():
+ item_id = "ff97dd87-a4a5-4a12-b412-cde99f33e00e"
+ data = {
+ "start_datetime": "2018-12-22T14:00:00+00:00",
+ "end_datetime": "2018-12-24T15:00:00+00:00",
+ "repeat_at": "15:30:00",
+ "process_after": 300,
+ }
+ expected_response = data.copy()
+ expected_response.update(
+ {
+ "start_process": "2018-12-22T14:05:00+00:00",
+ "duration": 176_100,
+ "item_id": item_id,
+ }
+ )
+ response = client.put(f"/items/{item_id}", json=data)
+ assert response.status_code == 200
+ assert response.json() == expected_response
projects / thirdparty / fastapi / fastapi.git / commitdiff
