aws-frauddetector-detector/src/aws_frauddetector_detector/helpers/model_helpers.py [400:555]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - def _get_variables_and_return_event_variables_model( frauddetector_client, variable_names, reference_variable_names: set ): collected_variables = [] for variable_name in variable_names: # use singular get_variables to preserve order (transient contract test bug workaround) get_variables_response = api_helpers.call_get_variables(frauddetector_client, variable_name) collected_variables.extend(get_variables_response.get("variables", [])) return _get_event_variables_model_for_given_variables( frauddetector_client, collected_variables, reference_variable_names ) def _get_event_variables_model_for_given_variables(frauddetector_client, variables, reference_variable_names: set): variable_models = [] for variable in variables: variable_tags = _get_tags_for_given_arn(frauddetector_client, variable.get("arn", "")) tag_models = get_tag_models_from_tags(variable_tags) variable_name = util.extract_name_from_arn(variable.get("arn", "")) LOG.debug(f"checking if {variable_name} is in {reference_variable_names}") if variable_name in reference_variable_names: LOG.debug(f"in reference set, {variable_name} is not inline") variable_model = models.EventVariable( Arn=variable.get("arn", ""), Name=variable_name, Tags=None, Description=None, DataType=None, DataSource=None, DefaultValue=None, VariableType=None, CreatedTime=None, LastUpdatedTime=None, Inline=False, ) else: LOG.debug(f"not in reference set, {variable_name} is inline") variable_model = models.EventVariable( Name=variable.get("name", ""), Tags=tag_models, Description=variable.get("description", ""), DataType=variable.get("dataType", ""), DataSource=variable.get("dataSource", ""), DefaultValue=variable.get("defaultValue", ""), VariableType=variable.get("variableType", ""), Arn=variable.get("arn", ""), CreatedTime=variable.get("createdTime", ""), LastUpdatedTime=variable.get("lastUpdatedTime", ""), Inline=True, ) # remove empty description/tags if not variable_model.Tags: del variable_model.Tags if variable_model.Description is None or variable_model.Description == "": del variable_model.Description variable_models.append(variable_model) return variable_models # Labels def _get_labels_and_return_labels_model(frauddetector_client, label_names, reference_label_names: set): label_models = [] for label_name in label_names: get_labels_response = api_helpers.call_get_labels(frauddetector_client, label_name) labels = get_labels_response.get("labels", []) if not labels: raise RuntimeError(f"Error! Expected an existing label, but label did not exist! label name {label_name}") label = labels[0] label_arn = label.get("arn", "") LOG.debug(f"checking if {label_name} is in {reference_label_names}") if label_name in reference_label_names: LOG.debug(f"in reference set, {label_name} is not inline") label_model = models.Label( Arn=label_arn, Name=label_name, Tags=None, Description=None, CreatedTime=None, LastUpdatedTime=None, Inline=False, ) else: LOG.debug(f"not in reference set, {label_name} is inline") label_tags = _get_tags_for_given_arn(frauddetector_client, label_arn) tag_models = get_tag_models_from_tags(label_tags) label_model = models.Label( Name=label_name, Tags=tag_models, Description=label.get("description", ""), Arn=label_arn, CreatedTime=label.get("createdTime", ""), LastUpdatedTime=label.get("lastUpdatedTime", ""), Inline=True, ) # remove empty description/tags if not label_model.Tags: del label_model.Tags if label_model.Description is None or label_model.Description == "": del label_model.Description label_models.append(label_model) return label_models # EntityTypes def _get_entity_types_and_return_entity_types_model( frauddetector_client, entity_type_names: List[str], reference_entity_type_names: set ) -> List[models.EntityType]: entity_type_models = [] for entity_type_name in entity_type_names: ( get_entity_types_worked, get_entity_types_response, ) = validation_helpers.check_if_get_entity_types_succeeds(frauddetector_client, entity_type_name) if not get_entity_types_worked: raise RuntimeError( f"Error! Expected an existing get entity type, " f"but entity type did not exist! entity type {entity_type_name}" ) entity_type = get_entity_types_response.get("entityTypes")[0] entity_type_arn = entity_type.get("arn", "") LOG.debug(f"checking if {entity_type_name} is in {reference_entity_type_names}") if entity_type_name in reference_entity_type_names: LOG.debug(f"in reference set, {entity_type_name} is not inline") entity_type_model = models.EntityType( Arn=entity_type_arn, Name=entity_type_name, Tags=None, Description=None, CreatedTime=None, LastUpdatedTime=None, Inline=False, ) else: LOG.debug(f"not in reference set, {entity_type_name} is inline") entity_type_tags = _get_tags_for_given_arn(frauddetector_client, entity_type.get("arn", "")) tag_models = get_tag_models_from_tags(entity_type_tags) entity_type_model = models.EntityType( Name=entity_type_name, Tags=tag_models, Description=entity_type.get("description", ""), Arn=entity_type_arn, CreatedTime=entity_type.get("createdTime", ""), LastUpdatedTime=entity_type.get("lastUpdatedTime", ""), Inline=True, ) # remove empty description/tags if not entity_type_model.Tags: del entity_type_model.Tags if entity_type_model.Description is None or entity_type_model.Description == "": del entity_type_model.Description entity_type_models.append(entity_type_model) return entity_type_models - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - aws-frauddetector-eventtype/src/aws_frauddetector_eventtype/helpers/model_helpers.py [114:269]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - def _get_variables_and_return_event_variables_model( frauddetector_client, variable_names, reference_variable_names: set ): collected_variables = [] for variable_name in variable_names: # use singular get_variables to preserve order (transient contract test bug workaround) get_variables_response = api_helpers.call_get_variables(frauddetector_client, variable_name) collected_variables.extend(get_variables_response.get("variables", [])) return _get_event_variables_model_for_given_variables( frauddetector_client, collected_variables, reference_variable_names ) def _get_event_variables_model_for_given_variables(frauddetector_client, variables, reference_variable_names: set): variable_models = [] for variable in variables: variable_tags = _get_tags_for_given_arn(frauddetector_client, variable.get("arn", "")) tag_models = get_tag_models_from_tags(variable_tags) variable_name = util.extract_name_from_arn(variable.get("arn", "")) LOG.debug(f"checking if {variable_name} is in {reference_variable_names}") if variable_name in reference_variable_names: LOG.debug(f"in reference set, {variable_name} is not inline") variable_model = models.EventVariable( Arn=variable.get("arn", ""), Name=variable_name, Tags=None, Description=None, DataType=None, DataSource=None, DefaultValue=None, VariableType=None, CreatedTime=None, LastUpdatedTime=None, Inline=False, ) else: LOG.debug(f"not in reference set, {variable_name} is inline") variable_model = models.EventVariable( Name=variable.get("name", ""), Tags=tag_models, Description=variable.get("description", ""), DataType=variable.get("dataType", ""), DataSource=variable.get("dataSource", ""), DefaultValue=variable.get("defaultValue", ""), VariableType=variable.get("variableType", ""), Arn=variable.get("arn", ""), CreatedTime=variable.get("createdTime", ""), LastUpdatedTime=variable.get("lastUpdatedTime", ""), Inline=True, ) # remove empty description/tags if not variable_model.Tags: del variable_model.Tags if variable_model.Description is None or variable_model.Description == "": del variable_model.Description variable_models.append(variable_model) return variable_models # Labels def _get_labels_and_return_labels_model(frauddetector_client, label_names, reference_label_names: set): label_models = [] for label_name in label_names: get_labels_response = api_helpers.call_get_labels(frauddetector_client, label_name) labels = get_labels_response.get("labels", []) if not labels: raise RuntimeError(f"Error! Expected an existing label, but label did not exist! label name {label_name}") label = labels[0] label_arn = label.get("arn", "") LOG.debug(f"checking if {label_name} is in {reference_label_names}") if label_name in reference_label_names: LOG.debug(f"in reference set, {label_name} is not inline") label_model = models.Label( Arn=label_arn, Name=label_name, Tags=None, Description=None, CreatedTime=None, LastUpdatedTime=None, Inline=False, ) else: LOG.debug(f"not in reference set, {label_name} is inline") label_tags = _get_tags_for_given_arn(frauddetector_client, label_arn) tag_models = get_tag_models_from_tags(label_tags) label_model = models.Label( Name=label_name, Tags=tag_models, Description=label.get("description", ""), Arn=label_arn, CreatedTime=label.get("createdTime", ""), LastUpdatedTime=label.get("lastUpdatedTime", ""), Inline=True, ) # remove empty description/tags if not label_model.Tags: del label_model.Tags if label_model.Description is None or label_model.Description == "": del label_model.Description label_models.append(label_model) return label_models # EntityTypes def _get_entity_types_and_return_entity_types_model( frauddetector_client, entity_type_names: List[str], reference_entity_type_names: set ) -> List[models.EntityType]: entity_type_models = [] for entity_type_name in entity_type_names: ( get_entity_types_worked, get_entity_types_response, ) = validation_helpers.check_if_get_entity_types_succeeds(frauddetector_client, entity_type_name) if not get_entity_types_worked: raise RuntimeError( f"Error! Expected an existing get entity type, " f"but entity type did not exist! entity type {entity_type_name}" ) entity_type = get_entity_types_response.get("entityTypes")[0] entity_type_arn = entity_type.get("arn", "") LOG.debug(f"checking if {entity_type_name} is in {reference_entity_type_names}") if entity_type_name in reference_entity_type_names: LOG.debug(f"in reference set, {entity_type_name} is not inline") entity_type_model = models.EntityType( Arn=entity_type_arn, Name=entity_type_name, Tags=None, Description=None, CreatedTime=None, LastUpdatedTime=None, Inline=False, ) else: LOG.debug(f"not in reference set, {entity_type_name} is inline") entity_type_tags = _get_tags_for_given_arn(frauddetector_client, entity_type.get("arn", "")) tag_models = get_tag_models_from_tags(entity_type_tags) entity_type_model = models.EntityType( Name=entity_type_name, Tags=tag_models, Description=entity_type.get("description", ""), Arn=entity_type_arn, CreatedTime=entity_type.get("createdTime", ""), LastUpdatedTime=entity_type.get("lastUpdatedTime", ""), Inline=True, ) # remove empty description/tags if not entity_type_model.Tags: del entity_type_model.Tags if entity_type_model.Description is None or entity_type_model.Description == "": del entity_type_model.Description entity_type_models.append(entity_type_model) return entity_type_models - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -