in src/lambdas/r8s_api_handler/processors/recommendation_history_processor.py [0:0]
def patch(self, event):
_LOG.debug(f'Update recommendation event: {event}')
validate_params(event, (INSTANCE_ID_ATTR, RECOMMENDATION_TYPE_ATTR,
FEEDBACK_STATUS_ATTR))
customer = event.get(PARAM_USER_CUSTOMER)
if customer == 'admin':
customer = event.get(CUSTOMER_ATTR)
if not customer:
_LOG.warning(f'\'{CUSTOMER_ATTR}\' must be '
f'specified for admin users.')
return build_response(
code=RESPONSE_BAD_REQUEST_CODE,
content=f'\'{CUSTOMER_ATTR}\' must be specified '
f'for admin users.'
)
instance_id = event.get(INSTANCE_ID_ATTR)
recommendation_type = event.get(RECOMMENDATION_TYPE_ATTR)
feedback_status = event.get(FEEDBACK_STATUS_ATTR)
self._validate_recommendation_type(
recommendation_type=recommendation_type)
self._validate_feedback_status(feedback_status=feedback_status,
recommendation_type=recommendation_type)
_LOG.debug(f'Searching for instance \'{instance_id}\' '
f'recommendation of type \'{recommendation_type}\'')
recommendation = list(
self.recommendation_history_service.get_recent_recommendation(
instance_id=instance_id,
recommendation_type=recommendation_type,
customer=customer,
limit=1
))
if not recommendation:
_LOG.error(f'No recommendations found matching given query.')
return build_response(
code=RESPONSE_RESOURCE_NOT_FOUND_CODE,
content=f'No recommendations found matching given query.'
)
recommendation = recommendation[0]
_LOG.debug(f'Updating recommendation')
recommendation = self.recommendation_history_service.save_feedback(
recommendation=recommendation,
feedback_status=feedback_status
)
_LOG.debug(f'Describing recommendation dto')
response = recommendation.get_dto()
_LOG.debug(f'Response: {response}')
return build_response(
code=RESPONSE_OK_CODE,
content=response
)