Module slack_bolt.lazy_listener

Lazy listener runner is a beta feature for the apps running on Function-as-a-Service platforms.

def respond_to_slack_within_3_seconds(body, ack):
    text = body.get("text")
    if text is None or len(text) == 0:
        ack(f":x: Usage: /start-process (description here)")
    else:
        ack(f"Accepted! (task: {body['text']})")

import time
def run_long_process(respond, body):
    time.sleep(5)  # longer than 3 seconds
    respond(f"Completed! (task: {body['text']})")

app.command("/start-process")(
    # ack() is still called within 3 seconds
    ack=respond_to_slack_within_3_seconds,
    # Lazy function is responsible for processing the event
    lazy=[run_long_process]
)

Refer to https://slack.dev/bolt-python/concepts#lazy-listeners for more details.

Expand source code 
"""Lazy listener runner is a beta feature for the apps running on Function-as-a-Service platforms.

    def respond_to_slack_within_3_seconds(body, ack):
        text = body.get("text")
        if text is None or len(text) == 0:
            ack(f":x: Usage: /start-process (description here)")
        else:
            ack(f"Accepted! (task: {body['text']})")

    import time
    def run_long_process(respond, body):
        time.sleep(5)  # longer than 3 seconds
        respond(f"Completed! (task: {body['text']})")

    app.command("/start-process")(
        # ack() is still called within 3 seconds
        ack=respond_to_slack_within_3_seconds,
        # Lazy function is responsible for processing the event
        lazy=[run_long_process]
    )

Refer to https://slack.dev/bolt-python/concepts#lazy-listeners for more details.
"""
# Don't add async module imports here
from .runner import LazyListenerRunner
from .thread_runner import ThreadLazyListenerRunner

__all__ = [
    "LazyListenerRunner",
    "ThreadLazyListenerRunner",
]

Sub-modules

slack_bolt.lazy_listener.async_internals
slack_bolt.lazy_listener.async_runner
slack_bolt.lazy_listener.asyncio_runner
slack_bolt.lazy_listener.internals
slack_bolt.lazy_listener.runner
slack_bolt.lazy_listener.thread_runner

Classes

class LazyListenerRunner
Expand source code 
class LazyListenerRunner(metaclass=ABCMeta):
    logger: Logger

    @abstractmethod
    def start(self, function: Callable[..., None], request: BoltRequest) -> None:
        """Starts a new lazy listener execution.

        Args:
            function: The function to run.
            request: The request to pass to the function. The object must be thread-safe.
        """
        raise NotImplementedError()

    def run(self, function: Callable[..., None], request: BoltRequest) -> None:
        """Synchronously runs the function with a given request data.

        Args:
            function: The function to run.
            request: The request to pass to the function. The object must be thread-safe.
        """
        build_runnable_function(
            func=function,
            logger=self.logger,
            request=request,
        )()

Subclasses

Class variables

var logger : logging.Logger

Methods

def run(self, function: Callable[..., None], request: BoltRequest) ‑> None

Synchronously runs the function with a given request data.

Args

function
The function to run.
request
The request to pass to the function. The object must be thread-safe.
Expand source code 
def run(self, function: Callable[..., None], request: BoltRequest) -> None:
    """Synchronously runs the function with a given request data.

    Args:
        function: The function to run.
        request: The request to pass to the function. The object must be thread-safe.
    """
    build_runnable_function(
        func=function,
        logger=self.logger,
        request=request,
    )()
def start(self, function: Callable[..., None], request: BoltRequest) ‑> None

Starts a new lazy listener execution.

Args

function
The function to run.
request
The request to pass to the function. The object must be thread-safe.
Expand source code 
@abstractmethod
def start(self, function: Callable[..., None], request: BoltRequest) -> None:
    """Starts a new lazy listener execution.

    Args:
        function: The function to run.
        request: The request to pass to the function. The object must be thread-safe.
    """
    raise NotImplementedError()
class ThreadLazyListenerRunner (logger: logging.Logger, executor: concurrent.futures._base.Executor)
Expand source code 
class ThreadLazyListenerRunner(LazyListenerRunner):
    logger: Logger

    def __init__(
        self,
        logger: Logger,
        executor: Executor,
    ):
        self.logger = logger
        self.executor = executor

    def start(self, function: Callable[..., None], request: BoltRequest) -> None:
        self.executor.submit(
            build_runnable_function(
                func=function,
                logger=self.logger,
                request=request,
            )
        )

Ancestors

Subclasses

Class variables

var logger : logging.Logger

Inherited members