in src/lambdas/r8s_api_handler/processors/shape_rule_processor.py [0:0]
def get(self, event):
_LOG.debug(f'Describe shape rule event: {event}')
_LOG.debug(f'Resolving applications')
applications = self.application_service.resolve_application(
event=event)
if not applications:
_LOG.warning(f'No application found matching given query.')
return build_response(
code=RESPONSE_BAD_REQUEST_CODE,
content=f'No application found matching given query.'
)
parent_id = event.get(PARENT_ID_ATTR)
app_ids = [app.application_id for app in applications]
parents = []
if parent_id:
_LOG.debug(f'Describing parent \'{parent_id}\'')
parent = self.get_parent(parent_id=parent_id,
application_ids=app_ids)
if parent:
self._validate_parent(parent=parent)
parents.append(parent)
else:
_LOG.debug(f'Describing parents from applications: '
f'f{", ".join(app_ids)}')
parents = self.get_parents(application_ids=app_ids)
if not parents:
_LOG.error(f'No parents found matching given query')
return build_response(
code=RESPONSE_BAD_REQUEST_CODE,
content=f'No parents found matching given query'
)
shape_rules: List[ShapeRule] = []
for parent in parents:
_LOG.debug(f'Resolving rules for parent '
f'\'{parent.parent_id}\'')
parent_meta = self.parent_service.get_parent_meta(
parent=parent)
parent_rules = self.parent_service.list_shape_rules(
parent_meta=parent_meta)
shape_rules.extend(parent_rules)
rule_id = event.get(ID_ATTR)
if rule_id:
_LOG.debug(f'Describing only rule with id \'{rule_id}\'')
shape_rules = [shape_rule for shape_rule in shape_rules
if shape_rule.rule_id == rule_id]
if not shape_rules:
_LOG.warning(f'No shape rules found matching given query.')
return build_response(
code=RESPONSE_RESOURCE_NOT_FOUND_CODE,
content=f'No shape rules found matching given query.'
)
shape_rule_dto = []
for shape_rule in shape_rules:
rule_dto = self.parent_service.get_shape_rule_dto(
shape_rule=shape_rule)
shape_rule_dto.append(rule_dto)
_LOG.debug(f'Response: {shape_rule_dto}')
return build_response(
code=RESPONSE_OK_CODE,
content=shape_rule_dto
)