in src/services/abs_lambda.py [0:0]
def __call__(self, event: dict, context: RequestContext
) -> tuple[ProcessedEvent, RequestContext]:
"""
Accepts API GW lambda proxy integration event
"""
if self._is_from_step_function(event):
_LOG.debug('Event came from step function. Preprocessing')
self._prepare_step_function_event(event)
body = event.get('body') or '{}'
if isinstance(body, str):
try:
body = self._decoder.decode(body)
except msgspec.ValidationError as e:
_LOG.warning('Invalid body type came. Returning 400')
raise ResponseFactory(HTTPStatus.BAD_REQUEST).message(
str(e)
).exc()
except msgspec.DecodeError as e:
_LOG.warning('Invalid incoming json. Returning 400')
raise ResponseFactory(HTTPStatus.BAD_REQUEST).message(
str(e)
).exc()
rc = event.get('requestContext') or {}
return {
'method': (method := HTTPMethod(event['httpMethod'])),
'resource': (res := CustodianEndpoint.match(rc['resourcePath'])),
'path': event['path'], # todo may be wrong if we use custom domain
'fullpath': rc['path'],
'cognito_username': deep_get(rc, ('authorizer', 'claims',
'cognito:username')),
'cognito_customer': (cst := deep_get(rc, ('authorizer', 'claims',
'custom:customer'))),
'cognito_user_id': deep_get(rc, ('authorizer', 'claims', 'sub')),
'cognito_user_role': deep_get(rc, ('authorizer', 'claims',
'custom:role')),
'permission': self._mapping.get((res, method)),
'is_system': cst == SYSTEM_CUSTOMER,
'body': body,
'query': dict(event.get('queryStringParameters') or {}),
'path_params': dict(event.get('pathParameters') or {}),
'tenant_access_payload': TenantsAccessPayload.build_denying_all(),
'additional_kwargs': dict(),
'headers': event['headers']
}, context