diff --git a/keyosk/resources/_shared.py b/keyosk/resources/_shared.py new file mode 100644 index 0000000..d555a50 --- /dev/null +++ b/keyosk/resources/_shared.py @@ -0,0 +1,190 @@ +import logging +from typing import Any +from typing import Dict +from typing import List +from typing import NamedTuple +from typing import Optional +from typing import Sequence +from typing import Tuple +from typing import Type +from typing import Union + +import flask +import flask_restful +import marshmallow as msh +import webargs + +from keyosk import __about__ +from keyosk import constants +from keyosk import exceptions + + +ResponseBody = Optional[Union[Dict[str, Any], List[Dict[str, Any]], List[str]]] + + +STATIC_HEADERS = {"x-keyosk-version": __about__.__version__} + + +class ResponseTuple(NamedTuple): + """Namedtuple representing the format of a flask-restful response tuple + + :param body: Response body; must be comprised only of JSON-friendly primative types + :param code: HTTP response code + :param headers: Dictionary of headers + """ + + body: ResponseBody + code: int + headers: Dict[str, str] + + +class PaginationData(NamedTuple): + """Tuple of data returned from the pagination function + :param data: Resulting paginated data to return in the response body + :param headers: Pagination headers to be included in the response headers + .. note:: The ``headers`` attribute **should not** include the default headers + """ + + data: Sequence[Any] + headers: Dict[str, str] + + +class KeyoskParser(webargs.flaskparser.FlaskParser): + + DEFAULT_VALIDATION_STATUS = 400 + + def handle_error(self, error: Type[Exception], *args, **kwargs): + logger = logging.getLogger(__name__) + logger.error(error) + raise error + + +def make_pagination(matches: Sequence[Any], request=flask.request) -> PaginationData: + """Paginate a sequence of result data + + Parse the pagination parameters out of the request body and filter the list of matches + accordinglly + :param matches: Sequence of results to paginate + :returns: Pagination data tuple with processed results, pagination parameters, and the headers + to return to the client + """ + + pagination_parameters = { + constants.HEADER_REQUEST_PAGE_OFFSET: webargs.fields.Integer( + location="headers", missing=0, validate=webargs.validate.Range(min=0) + ), + constants.HEADER_REQUEST_PAGE_LIMIT: webargs.fields.Integer( + location="headers", missing=0, validate=webargs.validate.Range(min=0) + ), + } + + try: + args = KeyoskParser().parse(pagination_parameters, request) + except webargs.ValidationError as err: + raise exceptions.InvalidPaginationParameterError( + "Invalid values specified for result pagination", + data={key: "; ".join(value) for key, value in err.messages.items()}, + ) from err + + try: + processed = matches[args[constants.HEADER_REQUEST_PAGE_OFFSET] :] + except IndexError: + raise exceptions.IndexOutOfRangeError( + f"Offset of '{args[constants.HEADER_REQUEST_PAGE_OFFSET]}' outside of resource range '0-{len(matches)}'", + data={ + constants.HEADER_RESPONSE_PAGE_TOTAL: len(matches), + constants.HEADER_RESPONSE_PAGE_LIMIT: args[ + constants.HEADER_REQUEST_PAGE_LIMIT + ], + constants.HEADER_RESPONSE_PAGE_OFFSET: args[ + constants.HEADER_REQUEST_PAGE_OFFSET + ], + }, + ) from None + + try: + processed = ( + processed[: args[constants.HEADER_REQUEST_PAGE_LIMIT]] + if args[constants.HEADER_REQUEST_PAGE_LIMIT] + else processed + ) + except IndexError: + # If an index error happens here, then it means that ``limit`` is greater than the remaining + # length of the list, after offset is applied. In this case we want to simply return + # whatever is left in the list, so we ignore the error and move on + pass + + return PaginationData( + data=processed, + headers={ + constants.HEADER_RESPONSE_PAGE_TOTAL: str(len(matches)), + constants.HEADER_RESPONSE_PAGE_OFFSET: str( + args[constants.HEADER_REQUEST_PAGE_OFFSET] + ), + constants.HEADER_RESPONSE_PAGE_LIMIT: str( + args[constants.HEADER_REQUEST_PAGE_LIMIT] + ), + }, + ) + + +class KeyoskResource(flask_restful.Resource): + """Extension of the default :class:`flask_restful.Resource` class + + Add a couple of useful things to the default resource class: + + * Adds the :meth:`options` method to respond to HTTP OPTION requests + * Type hints the :attr:`ROUTES` and :attr:`ARGS` attributes for usage in subclasses + + .. warning:: This class is a stub and should not be directly attached to an application + + :attribute ROUTES: Tuple of route paths that this resource should handle; can be unpacked into + ``flask_restful.Api().add_route()`` + :attribute ARGS: Dictionary mapping methods to their arguments for parsing + """ + + ROUTES: Tuple[str] + + ARGS: Dict[str, Union[msh.Schema, Dict[str, webargs.fields.Field]]] + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.logger = logging.getLogger(__name__) + + # pylint: disable=unused-argument,no-self-use + def options(self, *args, **kwargs) -> ResponseTuple: + """Give the supported HTTP verbs and request paramaeters for the resource""" + + verbs = ",".join([verb.upper() for verb in flask.request.url_rule.method]) + + return ResponseTuple( + body=None, code=204, headers={**{"Allowed": verbs}, **STATIC_HEADERS,}, + ) + + def respond(self, data: ResponseBody, code: int = 200): + """Create a response tuple from the current context + Helper function for generating defaults, parsing common data, and formatting the response. + :param data: Response data to return from the request + :param code: Response code to return; defaults to ``200: Ok`` + :param request: Request object to use for the request context; defaults to the current flask + request context + :returns: Response tuple ready to be returned out of a resource method + .. note:: This function will handle pagination and header assembly internally. The response data + passed to the ``data`` parameter should be unpaginated. + """ + + if isinstance(data, list): + pagination = make_pagination(data, flask.request) + body = pagination.data + else: + pagination = None + body = data + + headers = {**STATIC_HEADERS, **(pagination.headers if pagination else {})} + + return ResponseTuple( + body=body if code != 204 else None, code=code, headers=headers + ) + + def _head(self, response: ResponseTuple) -> ResponseTuple: + return ResponseTuple(body=None, code=response.code, headers=response.headers)