def create_transition_route_from_dataframe()

in src/dfcx_scrapi/tools/dataframe_functions.py [0:0]

• 38 lines of code
• 7 McCabe index (conditional complexity)

    def create_transition_route_from_dataframe(self, route_df):
        """Create transition route.

        Args:
          route_df: dataframe with data for a single route. Should only be one
          row. The columns required are:
            intent: intent id
            condition: string condition. ex.
              $session.params.dtmf_diy_opt_in = 1 AND
              $session.params.dtmf_2_techinternet = 2
            target_page: page id
            target_flow: flow id
            webhook: webhook id
            webhook_tag: string webhook tag
            custom_payload: a singular payload or list of payloads ex. [{}, {}]
            fulfillment_text: = list of text ["yo", "hi"]
            parameter_presets: = dictionary of parameter presets ex.
              {"param1":"value","param2":"othervalues"}
            rate_limiter: seconds to sleep between operations.

        Returns:
          The created transition route protobuf object
        """

        transition_route = types.TransitionRoute()

        route_dict = route_df.to_dict()
        transition_route.intent = route_dict.get("intent", None)
        transition_route.condition = route_dict.get("condition", None)
        transition_route.target_page = route_dict.get("target_page", None)
        transition_route.target_flow = route_dict.get("target_flow", None)

        # fulfillment
        fulfillment = types.Fulfillment()
        fulfillment.webhook = route_dict.get("webhook", None)
        fulfillment.tag = route_dict.get("webhook_tag", None)

        custom_payload = route_dict.get("custom_payload", None)
        custom_payload_list = []
        if custom_payload:
            custom_payload = json.loads(custom_payload)
            if ~isinstance(custom_payload, list):
                custom_payload = [custom_payload]
            for single_payload in custom_payload:
                custom_payload_list.append({"payload": single_payload})

        fulfillment_text = route_dict.get("fulfillment_text", None)

        # custom payloads and text
        payload = {
            "messages": custom_payload_list
            + [{"text": {"text": fulfillment_text}}]
        }

        payload_json = json.dumps(payload)
        payload_json = json.dumps(payload)
        fulfillment = types.Fulfillment.from_json(payload_json)

        # parameter - presets
        set_param_actions = []
        parameter_presets = route_dict.get("parameter_presets", None)
        if parameter_presets:
            parameter_presets = json.loads(parameter_presets)
            for param in parameter_presets.keys():
                set_param_action = types.Fulfillment.SetParameterAction()
                set_param_action.parameter = param
                set_param_action.value = parameter_presets[param]
                set_param_actions.append(set_param_action)
        fulfillment.set_parameter_actions = set_param_actions
        transition_route.trigger_fulfillment = fulfillment

        return transition_route