123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423 |
- # -*- coding: utf-8 -*-
- import json
- import typing
- from http import HTTPStatus
-
- import functools
-
- import bottle
-
- # TODO: Gérer les erreurs de schema
- # TODO: Gérer les cas ou c'est une liste la réponse (items, item_nb)
-
-
- # CHANGE
- from hapic.exception import InputValidationException, \
- OutputValidationException, InputWorkflowException, ProcessException
-
- flatten = lambda l: [item for sublist in l for item in sublist]
-
-
- _waiting = {}
- _endpoints = {}
- _default_global_context = None
- _default_global_error_schema = None
-
-
- def error_schema(schema):
- global _default_global_error_schema
- _default_global_error_schema = schema
-
- def decorator(func):
- @functools.wraps(func)
- def wrapper(*args, **kwargs):
- return func(*args, **kwargs)
- return wrapper
- return decorator
-
-
- def set_fake_default_context(context):
- global _default_global_context
- _default_global_context = context
-
-
- def _register(func):
- assert func not in _endpoints
- global _waiting
-
- _endpoints[func] = _waiting
- _waiting = {}
-
-
- def with_api_doc():
- def decorator(func):
-
- @functools.wraps(func)
- def wrapper(*args, **kwargs):
- return func(*args, **kwargs)
-
- _register(wrapper)
- return wrapper
-
- return decorator
-
-
- def with_api_doc_bis():
- def decorator(func):
-
- @functools.wraps(func)
- def wrapper(*args, **kwargs):
- return func(*args, **kwargs)
-
- _register(func)
- return wrapper
-
- return decorator
-
-
- def generate_doc(app=None):
- # TODO @Damien bottle specific code !
- app = app or bottle.default_app()
-
- route_by_callbacks = []
- routes = flatten(app.router.dyna_routes.values())
- for path, path_regex, route, func_ in routes:
- route_by_callbacks.append(route.callback)
-
- for func, descriptions in _endpoints.items():
- routes = flatten(app.router.dyna_routes.values())
- for path, path_regex, route, func_ in routes:
- if route.callback == func:
- print(route.method, path, descriptions)
- continue
-
-
- class RequestParameters(object):
- def __init__(
- self,
- path_parameters,
- query_parameters,
- body_parameters,
- form_parameters,
- header_parameters,
- ):
- self.path_parameters = path_parameters
- self.query_parameters = query_parameters
- self.body_parameters = body_parameters
- self.form_parameters = form_parameters
- self.header_parameters = header_parameters
-
-
- class ProcessValidationError(object):
- def __init__(
- self,
- error_message: str,
- error_details: dict,
- ) -> None:
- self.error_message = error_message
- self.error_details = error_details
-
-
- class ContextInterface(object):
- def get_request_parameters(self, *args, **kwargs) -> RequestParameters:
- raise NotImplementedError()
-
- def get_response(
- self,
- response: dict,
- http_code: int,
- ) -> typing.Any:
- raise NotImplementedError()
-
- def get_validation_error_response(
- self,
- error: ProcessValidationError,
- http_code: HTTPStatus=HTTPStatus.BAD_REQUEST,
- ) -> typing.Any:
- raise NotImplementedError()
-
-
- class BottleContext(ContextInterface):
- def get_request_parameters(self, *args, **kwargs) -> RequestParameters:
- return RequestParameters(
- path_parameters=bottle.request.url_args,
- query_parameters=bottle.request.params,
- body_parameters=bottle.request.json,
- form_parameters=bottle.request.forms,
- header_parameters=bottle.request.headers,
- )
-
- 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:
- 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),
- )
-
-
- class OutputProcessorInterface(object):
- def __init__(self):
- self.schema = None
-
- def process(self, value):
- raise NotImplementedError
-
- def get_validation_error(
- self,
- request_context: RequestParameters,
- ) -> ProcessValidationError:
- raise NotImplementedError
-
-
- class InputProcessorInterface(object):
- def __init__(self):
- self.schema = None
-
- def process(
- self,
- request_context: RequestParameters,
- ) -> typing.Any:
- raise NotImplementedError
-
- def get_validation_error(
- self,
- request_context: RequestParameters,
- ) -> ProcessValidationError:
- raise NotImplementedError
-
-
- class MarshmallowOutputProcessor(OutputProcessorInterface):
- def process(self, data: typing.Any):
- unmarshall = self.schema.dump(data)
- if unmarshall.errors:
- raise InputValidationException(
- 'Error when validate input: {}'.format(
- str(unmarshall.errors),
- )
- )
-
- return unmarshall.data
-
- def get_validation_error(self, data: dict) -> ProcessValidationError:
- marshmallow_errors = self.schema.dump(data).errors
- return ProcessValidationError(
- error_message='Validation error of output data',
- error_details=marshmallow_errors,
- )
-
-
- class MarshmallowInputProcessor(OutputProcessorInterface):
- def process(self, data: dict):
- unmarshall = self.schema.load(data)
- if unmarshall.errors:
- raise OutputValidationException(
- 'Error when validate ouput: {}'.format(
- str(unmarshall.errors),
- )
- )
-
- return unmarshall.data
-
- def get_validation_error(self, data: dict) -> ProcessValidationError:
- marshmallow_errors = self.schema.load(data).errors
- return ProcessValidationError(
- error_message='Validation error of input data',
- error_details=marshmallow_errors,
- )
-
-
- class HapicData(object):
- def __init__(self):
- self.body = {}
- self.path = {}
- self.query = {}
- self.headers = {}
-
-
- # TODO: Il faut un output_body et un output_header
- def output(
- schema,
- processor: OutputProcessorInterface=None,
- context: ContextInterface=None,
- default_http_code=200,
- default_error_code=500,
- ):
- processor = processor or MarshmallowOutputProcessor()
- processor.schema = schema
- context = context or _default_global_context
-
- def decorator(func):
- # @functools.wraps(func)
- def wrapper(*args, **kwargs):
- raw_response = func(*args, **kwargs)
- processed_response = processor.process(raw_response)
- prepared_response = context.get_response(
- processed_response,
- default_http_code,
- )
- return prepared_response
-
- _waiting['output'] = schema
-
- return wrapper
- return decorator
-
- # TODO: raccourcis 'input' tout court ?
- def input_body(
- schema,
- processor: InputProcessorInterface=None,
- context: ContextInterface=None,
- error_http_code: HTTPStatus=HTTPStatus.BAD_REQUEST,
- ):
- processor = processor or MarshmallowInputProcessor()
- processor.schema = schema
- context = context or _default_global_context
-
- def decorator(func):
-
- # @functools.wraps(func)
- def wrapper(*args, **kwargs):
- updated_kwargs = {'hapic_data': HapicData()}
- updated_kwargs.update(kwargs)
- hapic_data = updated_kwargs['hapic_data']
-
- request_parameters = context.get_request_parameters(
- *args,
- **updated_kwargs
- )
-
- try:
- hapic_data.body = processor.process(
- request_parameters.body_parameters,
- )
- except ProcessException:
- error = processor.get_validation_error(
- request_parameters.body_parameters,
- )
- error_response = context.get_validation_error_response(
- error,
- http_code=error_http_code,
- )
- return error_response
-
- return func(*args, **updated_kwargs)
-
- _waiting.setdefault('input', []).append(schema)
-
- return wrapper
- return decorator
-
-
- def input_path(
- schema,
- processor: InputProcessorInterface=None,
- context: ContextInterface=None,
- error_http_code=400,
- ):
- processor = processor or MarshmallowInputProcessor()
- processor.schema = schema
- context = context or _default_global_context
-
- def decorator(func):
-
- # @functools.wraps(func)
- def wrapper(*args, **kwargs):
- updated_kwargs = {'hapic_data': HapicData()}
- updated_kwargs.update(kwargs)
- hapic_data = updated_kwargs['hapic_data']
-
- request_parameters = context.get_request_parameters(*args, **updated_kwargs)
- hapic_data.path = processor.process(request_parameters.path_parameters)
-
- return func(*args, **updated_kwargs)
-
- _waiting.setdefault('input', []).append(schema)
-
- return wrapper
- return decorator
-
-
- def input_query(
- schema,
- processor: InputProcessorInterface=None,
- context: ContextInterface=None,
- error_http_code=400,
- ):
- processor = processor or MarshmallowInputProcessor()
- processor.schema = schema
- context = context or _default_global_context
-
- def decorator(func):
-
- # @functools.wraps(func)
- def wrapper(*args, **kwargs):
- updated_kwargs = {'hapic_data': HapicData()}
- updated_kwargs.update(kwargs)
- hapic_data = updated_kwargs['hapic_data']
-
- request_parameters = context.get_request_parameters(*args, **updated_kwargs)
- hapic_data.query = processor.process(request_parameters.query_parameters)
-
- return func(*args, **updated_kwargs)
-
- _waiting.setdefault('input', []).append(schema)
-
- return wrapper
- return decorator
-
-
- def input_headers(
- schema,
- processor: InputProcessorInterface,
- context: ContextInterface=None,
- error_http_code=400,
- ):
- processor = processor or MarshmallowInputProcessor()
- processor.schema = schema
- context = context or _default_global_context
-
- def decorator(func):
-
- # @functools.wraps(func)
- def wrapper(*args, **kwargs):
- updated_kwargs = {'hapic_data': HapicData()}
- updated_kwargs.update(kwargs)
- hapic_data = updated_kwargs['hapic_data']
-
- request_parameters = context.get_request_parameters(*args, **updated_kwargs)
- hapic_data.headers = processor.process(request_parameters.header_parameters)
-
- return func(*args, **updated_kwargs)
-
- _waiting.setdefault('input', []).append(schema)
-
- return wrapper
- return decorator
|