Introduce ListenerStartHandler in the listener runner for better managing Django DB connections (#514)

Author: seratch

slack_bolt/adapter/django/handler.py

@@ -9,6 +9,10 @@
 from slack_bolt.error import BoltError
 from slack_bolt.lazy_listener import ThreadLazyListenerRunner
 from slack_bolt.lazy_listener.internals import build_runnable_function
+from slack_bolt.listener.listener_start_handler import (
+    ListenerStartHandler,
+    DefaultListenerStartHandler,
+)
 from slack_bolt.listener.listener_completion_handler import (
     ListenerCompletionHandler,
     DefaultListenerCompletionHandler,
@@ -67,6 +71,16 @@ def release_thread_local_connections(logger: Logger, execution_timing: str):
         )
 
 
+class DjangoListenerStartHandler(ListenerStartHandler):
+    """Django sets DB connections as a thread-local variable per thread.
+    If the thread is not managed on the Django app side, the connections won't be released by Django.
+    This handler releases the connections every time a ThreadListenerRunner execution completes.
+    """
+
+    def handle(self, request: BoltRequest, response: Optional[BoltResponse]) -> None:
+        release_thread_local_connections(request.context.logger, "listener-start")
+
+
 class DjangoListenerCompletionHandler(ListenerCompletionHandler):
     """Django sets DB connections as a thread-local variable per thread.
     If the thread is not managed on the Django app side, the connections won't be released by Django.
@@ -86,6 +100,9 @@ def start(self, function: Callable[..., None], request: BoltRequest) -> None:
         )
 
         def wrapped_func():
+            release_thread_local_connections(
+                request.context.logger, "before-lazy-listener"
+            )
             try:
                 func()
             finally:
@@ -117,6 +134,29 @@ def __init__(self, app: App):  # type: ignore
             self.app.logger.debug("App.process_before_response is set to True")
             return
 
+        current_start_handler = listener_runner.listener_start_handler
+        if current_start_handler is not None and not isinstance(
+            current_start_handler, DefaultListenerStartHandler
+        ):
+            # As we run release_thread_local_connections() before listener executions,
+            # it's okay to skip calling the same connection clean-up method at the listener completion.
+            message = """As you've already set app.listener_runner.listener_start_handler to your own one,
+            Bolt skipped to set it to slack_sdk.adapter.django.DjangoListenerStartHandler.
+            
+            If you go with your own handler here, we highly recommend having the following lines of code 
+            in your handle() method to clean up unmanaged stale/old database connections:
+
+            from django.db import close_old_connections
+            close_old_connections()
+            """
+            self.app.logger.info(message)
+        else:
+            # for proper management of thread-local Django DB connections
+            self.app.listener_runner.listener_start_handler = (
+                DjangoListenerStartHandler()
+            )
+            self.app.logger.debug("DjangoListenerStartHandler has been enabled")
+
         current_completion_handler = listener_runner.listener_completion_handler
         if current_completion_handler is not None and not isinstance(
             current_completion_handler, DefaultListenerCompletionHandler
@@ -145,13 +185,6 @@ def handle(self, req: HttpRequest) -> HttpResponse:
                     bolt_resp = oauth_flow.handle_callback(to_bolt_request(req))
                     return to_django_response(bolt_resp)
         elif req.method == "POST":
-            # As bolt-python utilizes threads for async `ack()` method execution,
-            # we have to manually clean old/stale Django ORM connections bound to the "unmanaged" threads
-            # Refer to https://github.com/slackapi/bolt-python/issues/280 for more details.
-            release_thread_local_connections(
-                self.app.logger, "before-listener-invocation"
-            )
-            # And then, run the App listener/lazy listener here
             bolt_resp: BoltResponse = self.app.dispatch(to_bolt_request(req))
             return to_django_response(bolt_resp)
 

slack_bolt/app/app.py

@@ -23,6 +23,7 @@
 from slack_bolt.listener.builtins import TokenRevocationListeners
 from slack_bolt.listener.custom_listener import CustomListener
 from slack_bolt.listener.listener import Listener
+from slack_bolt.listener.listener_start_handler import DefaultListenerStartHandler
 from slack_bolt.listener.listener_completion_handler import (
     DefaultListenerCompletionHandler,
 )
@@ -317,6 +318,9 @@ def message_hello(message, say):
             listener_error_handler=DefaultListenerErrorHandler(
                 logger=self._framework_logger
             ),
+            listener_start_handler=DefaultListenerStartHandler(
+                logger=self._framework_logger
+            ),
             listener_completion_handler=DefaultListenerCompletionHandler(
                 logger=self._framework_logger
             ),

slack_bolt/app/async_app.py

@@ -8,6 +8,9 @@
 
 from slack_bolt.app.async_server import AsyncSlackAppServer
 from slack_bolt.listener.async_builtins import AsyncTokenRevocationListeners
+from slack_bolt.listener.async_listener_start_handler import (
+    AsyncDefaultListenerStartHandler,
+)
 from slack_bolt.listener.async_listener_completion_handler import (
     AsyncDefaultListenerCompletionHandler,
 )
@@ -337,6 +340,9 @@ async def message_hello(message, say):  # async function
             listener_error_handler=AsyncDefaultListenerErrorHandler(
                 logger=self._framework_logger
             ),
+            listener_start_handler=AsyncDefaultListenerStartHandler(
+                logger=self._framework_logger
+            ),
             listener_completion_handler=AsyncDefaultListenerCompletionHandler(
                 logger=self._framework_logger
             ),

slack_bolt/listener/async_listener_completion_handler.py

@@ -15,10 +15,9 @@ async def handle(
         request: AsyncBoltRequest,
         response: Optional[BoltResponse],
     ) -> None:
-        """Handles an unhandled exception.
+        """Do something extra after the listener execution
 
         Args:
-            error: The raised exception.
             request: The request.
             response: The response.
         """

slack_bolt/listener/async_listener_start_handler.py

@@ -0,0 +1,57 @@
+import inspect
+from abc import ABCMeta, abstractmethod
+from logging import Logger
+from typing import Callable, Dict, Any, Awaitable, Optional
+
+from slack_bolt.kwargs_injection.async_utils import build_async_required_kwargs
+from slack_bolt.request.async_request import AsyncBoltRequest
+from slack_bolt.response import BoltResponse
+
+
+class AsyncListenerStartHandler(metaclass=ABCMeta):
+    @abstractmethod
+    async def handle(
+        self,
+        request: AsyncBoltRequest,
+        response: Optional[BoltResponse],
+    ) -> None:
+        """Do something extra before the listener execution
+
+        Args:
+            request: The request.
+            response: The response.
+        """
+        raise NotImplementedError()
+
+
+class AsyncCustomListenerStartHandler(AsyncListenerStartHandler):
+    def __init__(self, logger: Logger, func: Callable[..., Awaitable[None]]):
+        self.func = func
+        self.logger = logger
+        self.arg_names = inspect.getfullargspec(func).args
+
+    async def handle(
+        self,
+        request: AsyncBoltRequest,
+        response: Optional[BoltResponse],
+    ) -> None:
+        kwargs: Dict[str, Any] = build_async_required_kwargs(
+            required_arg_names=self.arg_names,
+            logger=self.logger,
+            request=request,
+            response=response,
+            next_keys_required=False,
+        )
+        await self.func(**kwargs)
+
+
+class AsyncDefaultListenerStartHandler(AsyncListenerStartHandler):
+    def __init__(self, logger: Logger):
+        self.logger = logger
+
+    async def handle(
+        self,
+        request: AsyncBoltRequest,
+        response: Optional[BoltResponse],
+    ):
+        pass

slack_bolt/listener/asyncio_runner.py

@@ -7,6 +7,9 @@
 from slack_bolt.context.ack.async_ack import AsyncAck
 from slack_bolt.lazy_listener.async_runner import AsyncLazyListenerRunner
 from slack_bolt.listener.async_listener import AsyncListener
+from slack_bolt.listener.async_listener_start_handler import (
+    AsyncListenerStartHandler,
+)
 from slack_bolt.listener.async_listener_completion_handler import (
     AsyncListenerCompletionHandler,
 )
@@ -25,6 +28,7 @@ class AsyncioListenerRunner:
     logger: Logger
     process_before_response: bool
     listener_error_handler: AsyncListenerErrorHandler
+    listener_start_handler: AsyncListenerStartHandler
     listener_completion_handler: AsyncListenerCompletionHandler
     lazy_listener_runner: AsyncLazyListenerRunner
 
@@ -33,12 +37,14 @@ def __init__(
         logger: Logger,
         process_before_response: bool,
         listener_error_handler: AsyncListenerErrorHandler,
+        listener_start_handler: AsyncListenerStartHandler,
         listener_completion_handler: AsyncListenerCompletionHandler,
         lazy_listener_runner: AsyncLazyListenerRunner,
     ):
         self.logger = logger
         self.process_before_response = process_before_response
         self.listener_error_handler = listener_error_handler
+        self.listener_start_handler = listener_start_handler
         self.listener_completion_handler = listener_completion_handler
         self.lazy_listener_runner = lazy_listener_runner
 
@@ -55,6 +61,9 @@ async def run(
         if self.process_before_response:
             if not request.lazy_only:
                 try:
+                    await self.listener_start_handler.handle(
+                        request=request, response=response
+                    )
                     returned_value = await listener.run_ack_function(
                         request=request, response=response
                     )
@@ -113,6 +122,9 @@ async def run_ack_function_asynchronously(
                     response: BoltResponse,
                 ):
                     try:
+                        await self.listener_start_handler.handle(
+                            request=request, response=response
+                        )
                         await listener.run_ack_function(
                             request=request, response=response
                         )

slack_bolt/listener/listener_completion_handler.py

@@ -15,7 +15,7 @@ def handle(
         request: BoltRequest,
         response: Optional[BoltResponse],
     ) -> None:
-        """Handles an unhandled exception.
+        """Do something extra after the listener execution
 
         Args:
             request: The request.

slack_bolt/listener/listener_start_handler.py

@@ -0,0 +1,61 @@
+import inspect
+from abc import ABCMeta, abstractmethod
+from logging import Logger
+from typing import Callable, Dict, Any, Optional
+
+from slack_bolt.kwargs_injection import build_required_kwargs
+from slack_bolt.request.request import BoltRequest
+from slack_bolt.response.response import BoltResponse
+
+
+class ListenerStartHandler(metaclass=ABCMeta):
+    @abstractmethod
+    def handle(
+        self,
+        request: BoltRequest,
+        response: Optional[BoltResponse],
+    ) -> None:
+        """Do something extra before the listener execution.
+
+        This handler is useful if a developer needs to maintain/clean up
+        thread-local resources such as Django ORM database connections
+        before a listener execution starts.
+
+        Args:
+            request: The request.
+            response: The response.
+        """
+        raise NotImplementedError()
+
+
+class CustomListenerStartHandler(ListenerStartHandler):
+    def __init__(self, logger: Logger, func: Callable[..., None]):
+        self.func = func
+        self.logger = logger
+        self.arg_names = inspect.getfullargspec(func).args
+
+    def handle(
+        self,
+        request: BoltRequest,
+        response: Optional[BoltResponse],
+    ):
+        kwargs: Dict[str, Any] = build_required_kwargs(
+            required_arg_names=self.arg_names,
+            logger=self.logger,
+            request=request,
+            response=response,
+            next_keys_required=False,
+        )
+        self.func(**kwargs)
+
+
+class DefaultListenerStartHandler(ListenerStartHandler):
+    def __init__(self, logger: Logger):
+        self.logger = logger
+
+    def handle(
+        self,
+        request: BoltRequest,
+        response: Optional[BoltResponse],
+    ):
+        pass

slack_bolt/listener/thread_runner.py

@@ -5,6 +5,7 @@
 
 from slack_bolt.lazy_listener import LazyListenerRunner
 from slack_bolt.listener import Listener
+from slack_bolt.listener.listener_start_handler import ListenerStartHandler
 from slack_bolt.listener.listener_completion_handler import ListenerCompletionHandler
 from slack_bolt.listener.listener_error_handler import ListenerErrorHandler
 from slack_bolt.logger.messages import (
@@ -21,6 +22,7 @@ class ThreadListenerRunner:
     logger: Logger
     process_before_response: bool
     listener_error_handler: ListenerErrorHandler
+    listener_start_handler: ListenerStartHandler
     listener_completion_handler: ListenerCompletionHandler
     listener_executor: Executor
     lazy_listener_runner: LazyListenerRunner
@@ -30,13 +32,15 @@ def __init__(
         logger: Logger,
         process_before_response: bool,
         listener_error_handler: ListenerErrorHandler,
+        listener_start_handler: ListenerStartHandler,
         listener_completion_handler: ListenerCompletionHandler,
         listener_executor: Executor,
         lazy_listener_runner: LazyListenerRunner,
     ):
         self.logger = logger
         self.process_before_response = process_before_response
         self.listener_error_handler = listener_error_handler
+        self.listener_start_handler = listener_start_handler
         self.listener_completion_handler = listener_completion_handler
         self.listener_executor = listener_executor
         self.lazy_listener_runner = lazy_listener_runner
@@ -54,6 +58,10 @@ def run(  # type: ignore
         if self.process_before_response:
             if not request.lazy_only:
                 try:
+                    self.listener_start_handler.handle(
+                        request=request,
+                        response=response,
+                    )
                     returned_value = listener.run_ack_function(
                         request=request, response=response
                     )
@@ -109,6 +117,10 @@ def run(  # type: ignore
                 def run_ack_function_asynchronously():
                     nonlocal ack, request, response
                     try:
+                        self.listener_start_handler.handle(
+                            request=request,
+                            response=response,
+                        )
                         listener.run_ack_function(request=request, response=response)
                     except Exception as e:
                         # The default response status code is 500 in this case.

slack_bolt/version.py

@@ -1,2 +1,2 @@
 """Check the latest version at https://pypi.org/project/slack-bolt/"""
-__version__ = "1.9.4"
+__version__ = "1.10.0a"