Module slack_bolt.listener

Listeners process an incoming request from Slack if the request's type or data structure matches the predefined conditions of the listener. Typically, a listener acknowledge requests from Slack, process the request data, and may send response back to Slack.

Sub-modules

slack_bolt.listener.async_builtins

slack_bolt.listener.async_listener

slack_bolt.listener.async_listener_completion_handler

slack_bolt.listener.async_listener_error_handler

slack_bolt.listener.async_listener_start_handler

slack_bolt.listener.asyncio_runner

slack_bolt.listener.builtins

slack_bolt.listener.custom_listener

slack_bolt.listener.listener

slack_bolt.listener.listener_completion_handler

slack_bolt.listener.listener_error_handler

slack_bolt.listener.listener_start_handler

slack_bolt.listener.thread_runner

Classes

class CustomListener (*, app_name: str, ack_function: Callable[..., Optional[BoltResponse]], lazy_functions: Sequence[Callable[..., None]], matchers: Sequence[ListenerMatcher], middleware: Sequence[Middleware], auto_acknowledgement: bool = False, base_logger: Optional[logging.Logger] = None)

Expand source code

class CustomListener(Listener):
    app_name: str
    ack_function: Callable[..., Optional[BoltResponse]]  # type: ignore[assignment]
    lazy_functions: Sequence[Callable[..., None]]
    matchers: Sequence[ListenerMatcher]
    middleware: Sequence[Middleware]
    auto_acknowledgement: bool
    arg_names: MutableSequence[str]
    logger: Logger

    def __init__(
        self,
        *,
        app_name: str,
        ack_function: Callable[..., Optional[BoltResponse]],
        lazy_functions: Sequence[Callable[..., None]],
        matchers: Sequence[ListenerMatcher],
        middleware: Sequence[Middleware],
        auto_acknowledgement: bool = False,
        base_logger: Optional[Logger] = None,
    ):
        self.app_name = app_name
        self.ack_function = ack_function
        self.lazy_functions = lazy_functions
        self.matchers = matchers
        self.middleware = middleware
        self.auto_acknowledgement = auto_acknowledgement
        self.arg_names = get_arg_names_of_callable(ack_function)
        self.logger = get_bolt_app_logger(app_name, self.ack_function, base_logger)

    def run_ack_function(
        self,
        *,
        request: BoltRequest,
        response: BoltResponse,
    ) -> Optional[BoltResponse]:
        return self.ack_function(
            **build_required_kwargs(
                logger=self.logger,
                required_arg_names=self.arg_names,
                request=request,
                response=response,
                this_func=self.ack_function,
            )
        )

Ancestors

• Listener

Class variables

var ack_function : Callable[..., Optional[BoltResponse]]

var app_name : str

var arg_names : MutableSequence[str]

var auto_acknowledgement : bool

var lazy_functions : Sequence[Callable[..., None]]

var logger : logging.Logger

var matchers : Sequence[ListenerMatcher]

var middleware : Sequence[Middleware]

Inherited members

• Listener:
  • run_ack_function
  • run_middleware

class Listener

Expand source code

class Listener(metaclass=ABCMeta):
    matchers: Sequence[ListenerMatcher]
    middleware: Sequence[Middleware]
    ack_function: Callable[..., BoltResponse]
    lazy_functions: Sequence[Callable[..., None]]
    auto_acknowledgement: bool

    def matches(
        self,
        *,
        req: BoltRequest,
        resp: BoltResponse,
    ) -> bool:
        is_matched: bool = False
        for matcher in self.matchers:
            is_matched = matcher.matches(req, resp)
            if not is_matched:
                return is_matched
        return is_matched

    def run_middleware(
        self,
        *,
        req: BoltRequest,
        resp: BoltResponse,
    ) -> Tuple[Optional[BoltResponse], bool]:
        """Runs a middleware.

        Args:
            req: The incoming request
            resp: The current response

        Returns:
            A tuple of the processed response and a flag indicating termination
        """
        for m in self.middleware:
            middleware_state = {"next_called": False}

            def next_():
                middleware_state["next_called"] = True

            resp = m.process(req=req, resp=resp, next=next_)  # type: ignore[assignment]
            if not middleware_state["next_called"]:
                # next() was not called in this middleware
                return (resp, True)
        return (resp, False)

    @abstractmethod
    def run_ack_function(self, *, request: BoltRequest, response: BoltResponse) -> Optional[BoltResponse]:
        """Runs all the registered middleware and then run the listener function.

        Args:
            request: The incoming request
            response: The current response

        Returns:
            The processed response
        """
        raise NotImplementedError()

Subclasses

• CustomListener

Class variables

var ack_function : Callable[..., BoltResponse]

var auto_acknowledgement : bool

var lazy_functions : Sequence[Callable[..., None]]

var matchers : Sequence[ListenerMatcher]

var middleware : Sequence[Middleware]

Methods

def matches(self, *, req: BoltRequest, resp: BoltResponse) ‑> bool

def run_ack_function(self, *, request: BoltRequest, response: BoltResponse) ‑> Optional[BoltResponse]

Runs all the registered middleware and then run the listener function.

Args

request

The incoming request

response

The current response

Returns

The processed response

def run_middleware(self, *, req: BoltRequest, resp: BoltResponse) ‑> Tuple[Optional[BoltResponse], bool]

Runs a middleware.

Args

req

The incoming request

resp

The current response

Returns

A tuple of the processed response and a flag indicating termination