def get_orchestrator_state_str()

in azure/durable_functions/models/TaskOrchestrationExecutor.py [0:0]

• 26 lines of code
• 6 McCabe index (conditional complexity)

    def get_orchestrator_state_str(self) -> str:
        """Obtain a JSON-formatted string representing the orchestration's state.

        Returns
        -------
        str
            String represented orchestration's state, payload to the extension

        Raises
        ------
        Exception
            When the orchestration's state represents an error. The exception's
            message contains in it the string representation of the orchestration's
            state
        """
        if(self.output is not None):
            try:
                # Attempt to serialize the output. If serialization fails, raise an
                # error indicating that the orchestration output is not serializable,
                # which is not permitted in durable Python functions.
                json.dumps(self.output)
            except Exception as e:
                self.output = None
                self.exception = e

        exception_str = None
        if self.exception is not None:
            exception_str = str(self.exception)
            if not exception_str:
                exception_str = str(type(self.exception))

        state = OrchestratorState(
            is_done=self.orchestration_invocation_succeeded,
            actions=self.context._actions,
            output=self.output,
            replay_schema=self.context._replay_schema,
            error=exception_str,
            custom_status=self.context.custom_status
        )

        if self.exception is not None:
            # Create formatted error, using out-of-proc error schema
            error_label = "\n\n$OutOfProcData$:"
            state_str = state.to_json_string()
            formatted_error = f"{self.exception}{error_label}{state_str}"

            # Raise exception, re-set stack to original location
            raise Exception(formatted_error) from self.exception
        return state.to_json_string()