def from_json()

in aws_emr_launch/constructs/step_functions/emr_launch_function.py [0:0]

• 30 lines of code
• 5 McCabe index (conditional complexity)

    def from_json(self, property_values: Dict[str, Any]) -> "EMRLaunchFunction":
        self._launch_function_name = property_values["LaunchFunctionName"]
        self._namespace = property_values["Namespace"]

        profile_parts = property_values["EMRProfile"].split("/")
        self._emr_profile = emr_profile.EMRProfile.from_stored_profile(
            self, "EMRProfile", profile_parts[1], profile_parts[0]
        )
        config_parts = property_values["ClusterConfiguration"].split("/")
        self._cluster_configuration = cluster_configuration.ClusterConfiguration.from_stored_configuration(
            self, "ClusterConfiguration", config_parts[1], config_parts[0]
        )

        self._cluster_name = property_values["ClusterName"]
        self._default_fail_if_cluster_running = property_values["DefaultFailIfClusterRunning"]

        topic = property_values.get("SuccessTopic", None)
        self._success_topic = sns.Topic.from_topic_arn(self, "SuccessTopic", topic) if topic is not None else None

        topic = property_values.get("FailureTopic", None)
        self._failure_topic = sns.Topic.from_topic_arn(self, "FailureTopic", topic) if topic is not None else None

        func = property_values.get("OverrideClusterConfigsLambda", None)
        self._override_cluster_configs_lambda = (
            aws_lambda.Function.from_function_arn(self, "OverrideClusterConfigsLambda", func)
            if func is not None
            else None
        )

        self._allowed_cluster_config_overrides = property_values.get("AllowedClusterConfigOverrides", None)
        self._description = property_values.get("Description", None)
        self._cluster_tags = [core.Tag(t["Key"], t["Value"]) for t in property_values["ClusterTags"]]

        state_machine = property_values["StateMachine"]
        self._state_machine = sfn.StateMachine.from_state_machine_arn(self, "StateMachine", state_machine)

        self._wait_for_cluster_start = property_values.get("WaitForClusterStart", None)
        return self