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[..., BoltResponse | None],
lazy_functions: Sequence[Callable[..., None]],
matchers: Sequence[ListenerMatcher],
middleware: Sequence[Middleware],
auto_acknowledgement: bool = False,
base_logger: logging.Logger | None = 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 app_name : str

The type of the None singleton.

var arg_names : MutableSequence[str]

The type of the None singleton.

var logger : logging.Logger

The type of the None singleton.

Inherited members

• Listener:
  • ack_function
  • auto_acknowledgement
  • lazy_functions
  • matchers
  • middleware
  • 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]

The type of the None singleton.

var auto_acknowledgement : bool

The type of the None singleton.

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

The type of the None singleton.

var matchers : Sequence[ListenerMatcher]

The type of the None singleton.

var middleware : Sequence[Middleware]

The type of the None singleton.

Methods

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

Expand source code

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_ack_function(self,
*,
request: BoltRequest,
response: BoltResponse) ‑> BoltResponse | None

Expand source code

@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()

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[BoltResponse | None, bool]

Expand source code

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)

Runs a middleware.

Args

req

The incoming request

resp

The current response

Returns

A tuple of the processed response and a flag indicating termination