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.

Expand source code

"""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.
"""

# Don't add async module imports here
from .custom_listener import CustomListener
from .listener import Listener

builtin_listener_classes = [
    CustomListener,
]
for cls in builtin_listener_classes:
    Listener.register(cls)

__all__ = [
    "CustomListener",
    "Listener",
    "builtin_listener_classes",
]

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]]
    lazy_functions: Sequence[Callable[..., None]]
    matchers: Sequence[ListenerMatcher]
    middleware: Sequence[Middleware]  # type: ignore
    auto_acknowledgement: bool
    arg_names: Sequence[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],  # type: ignore
        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 : Sequence[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]  # type: ignore
    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_)
            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

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) ‑> 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

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

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

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_)
        if not middleware_state["next_called"]:
            # next() was not called in this middleware
            return (resp, True)
    return (resp, False)