# -*- coding: utf-8 -*- import json import re import typing from http import HTTPStatus import bottle from multidict import MultiDict from hapic.context import ContextInterface from hapic.context import RouteRepresentation from hapic.decorator import DecoratedController from hapic.decorator import DECORATION_ATTRIBUTE_NAME from hapic.exception import OutputValidationException from hapic.exception import NoRoutesException from hapic.exception import RouteNotFound from hapic.processor import RequestParameters from hapic.processor import ProcessValidationError # Bottle regular expression to locate url parameters BOTTLE_RE_PATH_URL = re.compile(r'<([^:<>]+)(?::[^<>]+)?>') class BottleContext(ContextInterface): def __init__(self, app: bottle.Bottle): self.app = app def get_request_parameters(self, *args, **kwargs) -> RequestParameters: path_parameters = dict(bottle.request.url_args) query_parameters = MultiDict(bottle.request.query.allitems()) body_parameters = dict(bottle.request.json or {}) form_parameters = MultiDict(bottle.request.forms.allitems()) header_parameters = dict(bottle.request.headers) files_parameters = dict(bottle.request.files) return RequestParameters( path_parameters=path_parameters, query_parameters=query_parameters, body_parameters=body_parameters, form_parameters=form_parameters, header_parameters=header_parameters, files_parameters=files_parameters, ) def get_response( self, response: dict, http_code: int, ) -> bottle.HTTPResponse: return bottle.HTTPResponse( body=json.dumps(response), headers=[ ('Content-Type', 'application/json'), ], status=http_code, ) def get_validation_error_response( self, error: ProcessValidationError, http_code: HTTPStatus=HTTPStatus.BAD_REQUEST, ) -> typing.Any: # TODO BS 20171010: Manage error schemas, see #4 from hapic.hapic import _default_global_error_schema unmarshall = _default_global_error_schema.dump(error) if unmarshall.errors: raise OutputValidationException( 'Validation error during dump of error response: {}'.format( str(unmarshall.errors) ) ) return bottle.HTTPResponse( body=json.dumps(unmarshall.data), headers=[ ('Content-Type', 'application/json'), ], status=int(http_code), ) def find_route( self, decorated_controller: DecoratedController, ) -> RouteRepresentation: if not self.app.routes: raise NoRoutesException('There is no routes in your bottle app') reference = decorated_controller.reference for route in self.app.routes: route_token = getattr( route.callback, DECORATION_ATTRIBUTE_NAME, None, ) match_with_wrapper = route.callback == reference.wrapper match_with_wrapped = route.callback == reference.wrapped match_with_token = route_token == reference.token if match_with_wrapper or match_with_wrapped or match_with_token: return RouteRepresentation( rule=self.get_swagger_path(route.rule), method=route.method.lower(), original_route_object=route, ) # TODO BS 20171010: Raise exception or print error ? see #10 raise RouteNotFound( 'Decorated route "{}" was not found in bottle routes'.format( decorated_controller.name, ) ) def get_swagger_path(self, contextualised_rule: str) -> str: return BOTTLE_RE_PATH_URL.sub(r'{\1}', contextualised_rule)