def execute_create_detector_handler_work()

in aws-frauddetector-detector/src/aws_frauddetector_detector/handler_workers.py [0:0]

• 25 lines of code
• 12 McCabe index (conditional complexity)

def execute_create_detector_handler_work(session: SessionProxy, model: models.ResourceModel, progress: ProgressEvent):
    afd_client = client_helpers.get_afd_client(session)

    # For contract_create_duplicate, we need to fail if the resource already exists
    get_detectors_works, _ = validation_helpers.check_if_get_detectors_succeeds(afd_client, model.DetectorId)
    if get_detectors_works:
        raise exceptions.AlreadyExists("detector", model.DetectorId)

    # For contract_invalid_create, fail if any read-only properties are present
    if model.Arn is not None or model.CreatedTime is not None or model.LastUpdatedTime is not None:
        raise exceptions.InvalidRequest("Error occurred: cannot create read-only properties.")

    # Validate existence of referenced resources, validate and create inline resources (except for Rules, Detector, DV)
    #   Also check existence of external models
    # TODO: split out creation from validation
    create_worker_helpers.validate_dependencies_for_detector_create(afd_client, model)

    # Create Detector, Rules, Detector Version ID
    model_helpers.put_detector_for_model(afd_client, model)
    rule_dicts = create_worker_helpers.create_rules_for_detector_resource(afd_client, model)
    detector_version_response = create_worker_helpers.create_detector_version_for_detector_resource(
        afd_client, model, rule_dicts
    )

    # The DV will be created as draft by default, so if the desired status is not draft, update DV status
    if model.DetectorVersionStatus != DRAFT_STATUS:
        api_helpers.call_update_detector_version_status(
            frauddetector_client=afd_client,
            detector_id=model.DetectorId,
            detector_version_id=detector_version_response.get("detectorVersionId", "1"),  # version here should be 1
            status=model.DetectorVersionStatus,
        )

    # after satisfying all contract tests and AFD requirements, get the resulting model
    model = read_worker_helpers.validate_detector_exists_and_return_detector_resource_model(afd_client, model)
    progress.resourceModel = model
    progress.status = OperationStatus.SUCCESS

    LOG.info(f"Returning Progress with status: {progress.status}")
    return progress