[WIP] feat(api): adds response formatting to api requests …

api/src/opentrons/app/models/item.py

@@ -0,0 +1,17 @@
+from opentrons.app.models.json_api.request import JsonApiRequest
+from pydantic import BaseModel
+from dataclasses import dataclass
+from uuid import uuid4
+
+
+@dataclass
+class ItemData:
+ name: str
+ quantity: int
+ price: float
+ id: str = str(uuid4().hex)
+
+class Item(BaseModel):
+ name: str
+ quantity: int
+ price: float

api/src/opentrons/app/models/json_api/errors.py

@@ -0,0 +1,52 @@
+from typing import Optional, List
+from pydantic import BaseModel, ValidationError
+
+from starlette.exceptions import HTTPException as StarletteHTTPException
+# https://github.com/encode/starlette/blob/master/starlette/status.py
+from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY
+
+from .filter import filter_none
+from .resource_links import ResourceLinks
+
+
+class ErrorSource(BaseModel):
+ pointer: Optional[str]
+ parameter: Optional[str]
+
+
+class Error(BaseModel):
+ """https://jsonapi.org/format/#error-objects"""
+ id: Optional[str]
+ links: Optional[ResourceLinks]
+ status: Optional[str]
+ code: Optional[str]
+ title: Optional[str]
+ detail: Optional[str]
+ source: Optional[ErrorSource]
+ meta: Optional[dict]
+
+
+class ErrorResponse(BaseModel):
+ errors: List[Error]
+
+def transform_to_json_api_errors(status_code, exception) -> dict:
+ if isinstance(exception, StarletteHTTPException):
+ request_error = {
+ 'status': status_code,
+ 'detail': exception.detail,
+ 'title': 'Bad Request'
+ }
+ error_response = ErrorResponse(errors=[request_error])
+ return filter_none(error_response.dict())
+ else:
+ def transform_error(error):
+ return {
+ 'status': status_code,
+ 'detail': error.get('msg'),
+ 'source': { 'pointer': '/' + '/'.join(error['loc']) },
+ 'title': error.get('type')
+ }
+ error_response = ErrorResponse(
+ errors=[transform_error(error) for error in exception.errors()]
+ )
+ return filter_none(error_response.dict())

api/src/opentrons/app/models/json_api/factory.py

@@ -0,0 +1,15 @@
+from typing import Any, Tuple
+
+from .request import JsonApiRequest, RequestModel
+from .response import JsonApiResponse, ResponseModel
+
+def JsonApiModel(
+ type_string: str,
+ attributes_model: Any,
+ *,
+ list_response: bool = False
+) -> Tuple[RequestModel, ResponseModel]:
+ return (
+ JsonApiRequest(type_string, attributes_model),
+ JsonApiResponse(type_string, attributes_model, use_list=list_response),
+ )

api/src/opentrons/app/models/json_api/filter.py

@@ -0,0 +1,17 @@
+from typing import TypeVar
+from collections.abc import Mapping, Iterable
+
+T = TypeVar('T')
+def filter_none(thing_to_traverse: T) -> T:
+ if isinstance(thing_to_traverse, dict):
+ return {
+ k: filter_none(v)
+ for k, v in thing_to_traverse.items()
+ if v is not None
+ }
+ elif isinstance(thing_to_traverse, list):
+ return [
+ filter_none(item)
+ for item in thing_to_traverse
+ ]
+ return thing_to_traverse

api/src/opentrons/app/models/json_api/request.py

@@ -0,0 +1,32 @@
+from typing import Generic, TypeVar, Optional, Any, Type
+from typing_extensions import Literal
+from pydantic.generics import GenericModel
+
+TypeT = TypeVar('TypeT')
+AttributesT = TypeVar('AttributesT')
+class RequestDataModel(GenericModel, Generic[TypeT, AttributesT]):
+ """
+ """
+ id: Optional[str]
+ type: TypeT
+ attributes: AttributesT
+
+
+DataT = TypeVar('DataT', bound=RequestDataModel)
+class RequestModel(GenericModel, Generic[DataT]):
+ """
+ """
+ data: DataT
+
+ def attributes(self):
+ return self.data.attributes
+
+def JsonApiRequest(type_string: str, attributes_model: Any) -> Type[RequestModel]:
+ request_data_model = RequestDataModel[
+ Literal[type_string],
+ attributes_model,
+ ]
+ request_data_model.__name__ = f'RequestData[{type_string}]'
+ request_model = RequestModel[request_data_model]
+ request_model.__name__ = f'Request[{type_string}]'
+ return request_model

api/src/opentrons/app/models/json_api/resource_links.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+class ResourceLinks(BaseModel):
+ self: str
+
+ResourceLinks.__doc__ = "https://jsonapi.org/format/#document-links"

api/src/opentrons/app/models/json_api/response.py

@@ -0,0 +1,76 @@
+from typing import Generic, TypeVar, Optional, List, Any, Type, get_type_hints
+from typing_extensions import Literal
+
+from pydantic.generics import GenericModel
+
+from .filter import filter_none
+from .resource_links import ResourceLinks
+
+TypeT = TypeVar('TypeT', bound=str)
+AttributesT = TypeVar('AttributesT')
+class ResponseDataModel(GenericModel, Generic[TypeT, AttributesT]):
+ """
+ """
+ id: str
+ type: TypeT
+ attributes: AttributesT = {}
+
+ class Config:
+ validate_all = True
+
+DataT = TypeVar('DataT', bound=ResponseDataModel)
+class ResponseModel(GenericModel, Generic[DataT]):
+ """
+ """
+ meta: Optional[dict]
+ data: DataT
+ links: Optional[ResourceLinks]
+
+ def dict(
+ self,
+ *,
+ serlialize_none: bool = False,
+ **kwargs
+ ):
+ response = super().dict(**kwargs)
+ if serlialize_none:
+ return response
+ return filter_none(response)
+
+ @classmethod
+ def resource_object(
+ cls,
+ *,
+ id: str,
+ attributes: Optional[dict] = None,
+ ) -> ResponseDataModel:
+ data_type = get_type_hints(cls)['data']
+ if getattr(data_type, '__origin__', None) is list:
+ data_type = data_type.__args__[0]
+ typename = get_type_hints(data_type)['type'].__args__[0]
+ return data_type(
+ id=id,
+ type=typename,
+ attributes=attributes or {},
+ )
+
+def JsonApiResponse(
+ type_string: str,
+ attributes_model: Any,
+ *,
+ use_list: bool = False
+) -> Type[ResponseModel]:
+ response_data_model = ResponseDataModel[
+ Literal[type_string],
+ attributes_model,
+ ]
+ if use_list:
+ response_data_model = List[response_data_model]
+ response_data_model.__name__ = f'ListResponseData[{type_string}]'
+ response_model = ResponseModel[response_data_model]
+ response_model.__name__ = f'ListResponse[{type_string}]'
+ else:
+ response_data_model.__name__ = f'ResponseData[{type_string}]'
+ response_model = ResponseModel[response_data_model]
+ response_model.__name__ = f'Response[{type_string}]'
+ return response_model

api/src/opentrons/app/routers/item.py

@@ -0,0 +1,51 @@
+import inspect
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import ValidationError
+
+from opentrons.app.models.item import Item, ItemData
+from opentrons.app.models.json_api.factory import JsonApiModel
+from opentrons.app.models.json_api.errors import ErrorResponse
+# https://github.com/encode/starlette/blob/master/starlette/status.py
+from starlette.status import HTTP_400_BAD_REQUEST, HTTP_422_UNPROCESSABLE_ENTITY
+
+router = APIRouter()
+
+ITEM_TYPE_NAME = "item"
+ItemRequest, ItemResponse = JsonApiModel(ITEM_TYPE_NAME, Item)
+
+@router.get("/items/{item_id}",
+ description="Get an individual item by it's ID",
+ summary="Get an individual item",
+ response_model=ItemResponse,
+ responses={
+ HTTP_422_UNPROCESSABLE_ENTITY: { "model": ErrorResponse },
+ })
+async def get_item(item_id: int) -> ItemResponse:
+ try:
+ # NOTE(isk: 3/10/20): mock DB / robot response
+ item = Item(name="apple", quantity=10, price=1.20)
+ data = ItemResponse.resource_object(id=item_id, attributes=item)
+ return ItemResponse(data=data, links={"self": f'/items/{item_id}'})
+ except ValidationError as e:
+ raise HTTPException(status_code=HTTP_422_UNPROCESSABLE_ENTITY, detail=e)
+
+@router.post("/items",
+ description="Create an item",
+ summary="Create an item via post route",
+ response_model=ItemResponse,
+ responses={
+ HTTP_400_BAD_REQUEST: { "model": ErrorResponse },
+ HTTP_422_UNPROCESSABLE_ENTITY: { "model": ErrorResponse },
+ })
+async def create_item(item_request: ItemRequest) -> ItemResponse:
+ try:
+ attributes = item_request.attributes().dict()
+ # NOTE(isk: 3/10/20): mock DB / robot response
+ item = ItemData(**attributes)
+ data = ItemResponse.resource_object(id=item.id, attributes=vars(item))
+ return ItemResponse(data=data, links={"self": f'/items/{item.id}'})
+ except ValidationError as e:
+ raise HTTPException(status_code=HTTP_422_UNPROCESSABLE_ENTITY, detail=e)
+ except Exception as e:
+ raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail=e)

api/tests/opentrons/app/helpers.py

@@ -0,0 +1,19 @@
+from pydantic import BaseModel
+from opentrons.app.models.json_api.request import JsonApiRequest
+from dataclasses import dataclass
+from uuid import uuid4
+
+@dataclass
+class ItemData:
+ name: str
+ quantity: int
+ price: float
+ id: str = str(uuid4().hex)
+
+class ItemModel(BaseModel):
+ name: str
+ quantity: int
+ price: float
+
+item_type_name = 'item'
+ItemRequest = JsonApiRequest(item_type_name, ItemModel)