slack_bolt.workflows.step API documentation

class Complete:
    """`complete()` utility to tell Slack the completion of a step from app execution.

        def execute(step, complete, fail):
            inputs = step["inputs"]
            # if everything was successful
            outputs = {
                "task_name": inputs["task_name"]["value"],
                "task_description": inputs["task_description"]["value"],
            }
            complete(outputs=outputs)

        ws = WorkflowStep(
            callback_id="add_task",
            edit=edit,
            save=save,
            execute=execute,
        )
        app.step(ws)

    This utility is a thin wrapper of workflows.stepCompleted API method.
    Refer to https://api.slack.com/methods/workflows.stepCompleted for details.
    """

    def __init__ (self, *, client: WebClient, body: dict):
        self.client = client
        self.body = body

    def __call__ (self, **kwargs) -> None:
        self.client.workflows_stepCompleted(
            workflow_step_execute_id=self.body["event"]["workflow_step"]["workflow_step_execute_id"],
            **kwargs,
        )

def execute(step, complete, fail):
    inputs = step["inputs"]
    # if everything was successful
    outputs = {
        "task_name": inputs["task_name"]["value"],
        "task_description": inputs["task_description"]["value"],
    }
    complete(outputs=outputs)

ws = WorkflowStep(
    callback_id="add_task",
    edit=edit,
    save=save,
    execute=execute,
)
app.step(ws)

class Configure:
    """`configure()` utility to send the modal view in Workflow Builder.

        def edit(ack, step, configure):
            ack()

            blocks = [
                {
                    "type": "input",
                    "block_id": "task_name_input",
                    "element": {
                        "type": "plain_text_input",
                        "action_id": "name",
                        "placeholder": {"type": "plain_text", "text": "Add a task name"},
                    },
                    "label": {"type": "plain_text", "text": "Task name"},
                },
            ]
            configure(blocks=blocks)

        ws = WorkflowStep(
            callback_id="add_task",
            edit=edit,
            save=save,
            execute=execute,
        )
        app.step(ws)

    Refer to https://docs.slack.dev/legacy/legacy-steps-from-apps/ for details.
    """

    def __init__ (self, *, callback_id: str, client: WebClient, body: dict):
        self.callback_id = callback_id
        self.client = client
        self.body = body

    def __call__ (self, *, blocks: Optional[Sequence[Union[dict, Block]]] = None, **kwargs) -> None:
        self.client.views_open(
            trigger_id=self.body["trigger_id"],
            view={
                "type": "workflow_step",
                "callback_id": self.callback_id,
                "blocks": blocks,
                **kwargs,
            },
        )

def edit(ack, step, configure):
    ack()

    blocks = [
        {
            "type": "input",
            "block_id": "task_name_input",
            "element": {
                "type": "plain_text_input",
                "action_id": "name",
                "placeholder": {"type": "plain_text", "text": "Add a task name"},
            },
            "label": {"type": "plain_text", "text": "Task name"},
        },
    ]
    configure(blocks=blocks)

ws = WorkflowStep(
    callback_id="add_task",
    edit=edit,
    save=save,
    execute=execute,
)
app.step(ws)

class Fail:
    """`fail()` utility to tell Slack the execution failure of a step from app.

        def execute(step, complete, fail):
            inputs = step["inputs"]
            # if something went wrong
            error = {"message": "Just testing step failure!"}
            fail(error=error)

        ws = WorkflowStep(
            callback_id="add_task",
            edit=edit,
            save=save,
            execute=execute,
        )
        app.step(ws)

    This utility is a thin wrapper of workflows.stepFailed API method.
    Refer to https://api.slack.com/methods/workflows.stepFailed for details.
    """

    def __init__ (self, *, client: WebClient, body: dict):
        self.client = client
        self.body = body

    def __call__ (
        self,
        *,
        error: dict,
    ) -> None:
        self.client.workflows_stepFailed(
            workflow_step_execute_id=self.body["event"]["workflow_step"]["workflow_step_execute_id"],
            error=error,
        )

def execute(step, complete, fail):
    inputs = step["inputs"]
    # if something went wrong
    error = {"message": "Just testing step failure!"}
    fail(error=error)

ws = WorkflowStep(
    callback_id="add_task",
    edit=edit,
    save=save,
    execute=execute,
)
app.step(ws)

class Update:
    """`update()` utility to tell Slack the processing results of a `save` listener.

        def save(ack, view, update):
            ack()

            values = view["state"]["values"]
            task_name = values["task_name_input"]["name"]
            task_description = values["task_description_input"]["description"]

            inputs = {
                "task_name": {"value": task_name["value"]},
                "task_description": {"value": task_description["value"]}
            }
            outputs = [
                {
                    "type": "text",
                    "name": "task_name",
                    "label": "Task name",
                },
                {
                    "type": "text",
                    "name": "task_description",
                    "label": "Task description",
                }
            ]
            update(inputs=inputs, outputs=outputs)

        ws = WorkflowStep(
            callback_id="add_task",
            edit=edit,
            save=save,
            execute=execute,
        )
        app.step(ws)

    This utility is a thin wrapper of workflows.stepFailed API method.
    Refer to https://api.slack.com/methods/workflows.updateStep for details.
    """

    def __init__ (self, *, client: WebClient, body: dict):
        self.client = client
        self.body = body

    def __call__ (self, **kwargs) -> None:
        self.client.workflows_updateStep(
            workflow_step_edit_id=self.body["workflow_step"]["workflow_step_edit_id"],
            **kwargs,
        )

def save(ack, view, update):
    ack()

    values = view["state"]["values"]
    task_name = values["task_name_input"]["name"]
    task_description = values["task_description_input"]["description"]

    inputs = {
        "task_name": {"value": task_name["value"]},
        "task_description": {"value": task_description["value"]}
    }
    outputs = [
        {
            "type": "text",
            "name": "task_name",
            "label": "Task name",
        },
        {
            "type": "text",
            "name": "task_description",
            "label": "Task description",
        }
    ]
    update(inputs=inputs, outputs=outputs)

ws = WorkflowStep(
    callback_id="add_task",
    edit=edit,
    save=save,
    execute=execute,
)
app.step(ws)

class WorkflowStep:
    callback_id: Union[str, Pattern]
    """The Callback ID of the step from app"""
    edit: Listener
    """`edit` listener, which displays a modal in Workflow Builder"""
    save: Listener
    """`save` listener, which accepts workflow creator's data submission in Workflow Builder"""
    execute: Listener
    """`execute` listener, which processes step from app execution"""

    def __init__ (
        self,
        *,
        callback_id: Union[str, Pattern],
        edit: Union[Callable[..., Optional[BoltResponse]], Listener, Sequence[Callable]],
        save: Union[Callable[..., Optional[BoltResponse]], Listener, Sequence[Callable]],
        execute: Union[Callable[..., Optional[BoltResponse]], Listener, Sequence[Callable]],
        app_name: Optional[str] = None,
        base_logger: Optional[Logger] = None,
    ):
        """
        Deprecated:
            Steps from apps for legacy workflows are now deprecated.
            Use new custom steps: https://docs.slack.dev/workflows/workflow-steps/

        Args:
            callback_id: The callback_id for this step from app
            edit: Either a single function or a list of functions for opening a modal in the builder UI
                When it's a list, the first one is responsible for ack() while the rest are lazy listeners.
            save: Either a single function or a list of functions for handling modal interactions in the builder UI
                When it's a list, the first one is responsible for ack() while the rest are lazy listeners.
            execute: Either a single function or a list of functions for handling step from app executions
                When it's a list, the first one is responsible for ack() while the rest are lazy listeners.
            app_name: The app name that can be mainly used for logging
            base_logger: The logger instance that can be used as a template when creating this step's logger
        """
        self.callback_id = callback_id
        app_name = app_name or __name__
        self.edit = self.build_listener(
            callback_id=callback_id,
            app_name=app_name,
            listener_or_functions=edit,
            name="edit",
            base_logger=base_logger,
        )
        self.save = self.build_listener(
            callback_id=callback_id,
            app_name=app_name,
            listener_or_functions=save,
            name="save",
            base_logger=base_logger,
        )
        self.execute = self.build_listener(
            callback_id=callback_id,
            app_name=app_name,
            listener_or_functions=execute,
            name="execute",
            base_logger=base_logger,
        )

    @classmethod
    def builder(cls, callback_id: Union[str, Pattern], base_logger: Optional[Logger] = None) -> WorkflowStepBuilder:
        """
        Deprecated:
            Steps from apps for legacy workflows are now deprecated.
            Use new custom steps: https://docs.slack.dev/workflows/workflow-steps/
        """
        return WorkflowStepBuilder(
            callback_id,
            base_logger=base_logger,
        )

    @classmethod
    def build_listener(
        cls,
        callback_id: Union[str, Pattern],
        app_name: str,
        listener_or_functions: Union[Listener, Callable, List[Callable]],
        name: str,
        matchers: Optional[List[ListenerMatcher]] = None,
        middleware: Optional[List[Middleware]] = None,
        base_logger: Optional[Logger] = None,
    ) -> Listener:
        if listener_or_functions is None:
            raise BoltError(f"{name} listener is required (callback_id: {callback_id})")

        if isinstance(listener_or_functions, Callable):
            listener_or_functions = [listener_or_functions]

        if isinstance(listener_or_functions, Listener):
            return listener_or_functions
        elif isinstance(listener_or_functions, list):
            matchers = matchers if matchers else []
            matchers.insert(
                0,
                cls._build_primary_matcher(
                    name,
                    callback_id,
                    base_logger=base_logger,
                ),
            )
            middleware = middleware if middleware else []
            middleware.insert(
                0,
                cls._build_single_middleware(
                    name,
                    callback_id,
                    base_logger=base_logger,
                ),
            )
            functions = listener_or_functions
            ack_function = functions.pop(0)
            return CustomListener(
                app_name=app_name,
                matchers=matchers,
                middleware=middleware,
                ack_function=ack_function,
                lazy_functions=functions,
                auto_acknowledgement=name == "execute",
                base_logger=base_logger,
            )
        else:
            raise BoltError(f"Invalid {name} listener: {type(listener_or_functions)} detected (callback_id: {callback_id})")

    @classmethod
    def _build_primary_matcher(
        cls,
        name: str,
        callback_id: Union[str, Pattern],
        base_logger: Optional[Logger] = None,
    ) -> ListenerMatcher:
        if name == "edit":
            return workflow_step_edit(callback_id, base_logger=base_logger)
        elif name == "save":
            return workflow_step_save(callback_id, base_logger=base_logger)
        elif name == "execute":
            return workflow_step_execute(callback_id, base_logger=base_logger)
        else:
            raise ValueError(f"Invalid name {name}")

    @classmethod
    def _build_single_middleware(
        cls,
        name: str,
        callback_id: Union[str, Pattern],
        base_logger: Optional[Logger] = None,
    ) -> Middleware:
        if name == "edit":
            return _build_edit_listener_middleware(callback_id, base_logger=base_logger)
        elif name == "save":
            return _build_save_listener_middleware(base_logger=base_logger)
        elif name == "execute":
            return _build_execute_listener_middleware(base_logger=base_logger)
        else:
            raise ValueError(f"Invalid name {name}")

class WorkflowStepMiddleware(Middleware):
    """Base middleware for step from app specific ones"""

    def __init__ (self, step: WorkflowStep):
        self.step = step

    def process(
        self,
        *,
        req: BoltRequest,
        resp: BoltResponse,
        # As this method is not supposed to be invoked by bolt-python users,
        # the naming conflict with the built-in one affects
        # only the internals of this method
        next: Callable[[], BoltResponse],
    ) -> Optional[BoltResponse]:

        if self.step.edit.matches(req=req, resp=resp):
            resp = self._run(self.step.edit, req, resp)
            if resp is not None:
                return resp
        elif self.step.save.matches(req=req, resp=resp):
            resp = self._run(self.step.save, req, resp)
            if resp is not None:
                return resp
        elif self.step.execute.matches(req=req, resp=resp):
            resp = self._run(self.step.execute, req, resp)
            if resp is not None:
                return resp

        return next()

    @staticmethod
    def _run(
        listener: Listener,
        req: BoltRequest,
        resp: BoltResponse,
    ) -> Optional[BoltResponse]:
        resp, next_was_not_called = listener.run_middleware(req=req, resp=resp)
        if next_was_not_called:
            return None

        return req.context.listener_runner.run(
            request=req,
            response=resp,
            listener_name=get_name_for_callable(listener.ack_function),
            listener=listener,
        )