def _callback()

in src/onprem/api/app.py [0:0]
47 lines of code
5 McCabe index (conditional complexity)

    def _callback(self, decoded_token: dict | None = None, **path_params):
        method = request.method
        path = request.route.rule
        ln = self._endpoint_to_lambda[(path, method)]
        handler = self.lambda_name_to_handler.get(ln)
        if not handler:
            resp = ResponseFactory(HTTPStatus.NOT_FOUND).default().build()
            return HTTPResponse(
                status=resp['statusCode'],
                body=resp['body'],
                headers=resp['headers']
            )
        event = {
            'httpMethod': request.method,
            'path': request.path,
            'headers': dict(request.headers),
            'requestContext': {
                'stage': self._dp_wrapper.stage,
                'resourcePath': path.replace('<', '{').replace('>', '}').replace('proxy', 'proxy+'),  # kludge
                'path': request.fullpath
            },
            'pathParameters': path_params
        }
        if decoded_token:
            event['requestContext']['authorizer'] = {
                'claims': {
                    'cognito:username': decoded_token.get('cognito:username'),
                    'sub': decoded_token.get('sub'),
                    'custom:customer': decoded_token.get(
                        'custom:customer'),
                    'custom:role': decoded_token.get('custom:role'),
                    'custom:tenants': decoded_token.get('custom:tenants') or ''
                }
            }

        if method == 'GET':
            event['queryStringParameters'] = dict(request.query)
        else:
            event['body'] = request.body.read().decode()
            event['isBase64Encoded'] = False

        _LOG.info(f'Handling request: {request.method}:{request.path}')
        response = handler(event, RequestContext())
        _LOG.info('Request was handled. Returning response')

        return HTTPResponse(
            body=response['body'],
            status=response['statusCode'],
            headers=response['headers']
        )