# Copyright 2017 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). # You may not use this file except in compliance with the License. # A copy of the License is located at # # http://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is distributed # on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either # express or implied. See the License for the specific language governing # permissions and limitations under the License. import decimal import functools import json import traceback import os class ClientError(Exception): status_code = 400 class BadRequestError(ClientError): pass class NotFoundError(ClientError): status_code = 404 class DecimalEncoder(json.JSONEncoder): def default(self, o): if isinstance(o, decimal.Decimal): return float(o) return super(DecimalEncoder, self).default(o) class Response: def __init__(self, body=None, headers=None, status_code=None): self.headers = headers or {} self.body = body self.status_code = status_code def to_response(self): response = {'headers': self.headers} status_code = self.status_code if self.body is not None: if status_code is None: status_code = 200 if 'Content-Type' in response['headers']: response['body'] = self.body else: response['headers']['Content-Type'] = 'application/json' response['body'] = json.dumps(self.body, cls=DecimalEncoder, sort_keys=True, indent=2) if status_code is None: status_code = 201 response['statusCode'] = status_code return response class route: # Shared route registry registry = {} def __init__(self, path, methods): self.path = path self.methods = methods def __call__(self, func): """ Helper for handling exceptions within the lambda function and returning back appropriate responses """ assert func.__name__ not in self.__class__.registry, f"There are 2 routed functions called {func.__name__}" @functools.wraps(func) def wrapper(event, context=None): try: if event['body'] is None or isinstance(event['body'], str): event['body'] = json.loads(event.get('body', None) or '{}') if not isinstance(event['body'], dict): raise Exception except Exception: return Response(body=f"Malformed JSON: {event['body']}", status_code=400).to_response() try: response = func(event) if not isinstance(response, Response): response = Response(body=response) return response.to_response() except ClientError as e: return Response(body=str(e), status_code=e.status_code).to_response() except Exception: return Response( body=f"Internal Service Exception: {traceback.format_exc()}", status_code=500 ).to_response() wrapper.route = self self.__class__.registry[func.__name__] = wrapper setattr(self.__class__, func.__name__, wrapper) return wrapper @route('/config', methods=['GET']) def config(event): return { 'USER_POOL_ID': os.environ.get('USER_POOL_ID', None), 'APP_CLIENT_ID': os.environ.get('APP_CLIENT_ID', None), }