bolster slack integration

Author: luke-beamcloud

examples/slackdemo.py

@@ -8,6 +8,7 @@
     SlackIntegration,
     SlackMessageEvent,
 )
+from fastloop.models import LoopEvent
 
 app = FastLoop(name="slackdemo")
 
@@ -16,27 +17,38 @@ class AppContext(LoopContext):
     client: Any
 
 
+async def resolve_bot_token(context: LoopContext, event: LoopEvent) -> str:
+    """
+    Resolve the bot token for this workspace.
+
+    In a real app, you would look this up from a database based on the team_id
+    that was stored when the workspace installed your Slack app via OAuth.
+
+    Example:
+        team_id = getattr(event, 'team', None)
+        return await db.get_bot_token(team_id)
+    """
+    return os.getenv("SLACK_BOT_TOKEN") or ""
+
+
 async def analyze_file(context: AppContext):
     file_shared: SlackFileSharedEvent | None = await context.wait_for(
         SlackFileSharedEvent, timeout=1
     )
     if not file_shared:
         return
 
-    file_bytes = await file_shared.download_file()
+    file_bytes = await file_shared.download_file(context)
     with open("something.png", "wb") as f:
         f.write(file_bytes)
 
 
 @app.loop(
     "filebot",
-    # start_event=SlackAppMentionEvent,
     integrations=[
         SlackIntegration(
-            app_id=os.getenv("SLACK_APP_ID") or "",
-            bot_token=os.getenv("SLACK_BOT_TOKEN") or "",
-            signing_secret=os.getenv("SLACK_SIGNING_SECRET") or "",
-            client_id=os.getenv("SLACK_CLIENT_ID") or "",
+            signing_secret=os.getenv("SLACK_SIGNING_SECRET"),
+            setup=resolve_bot_token,
         )
     ],
 )

fastloop/constants.py

@@ -6,6 +6,6 @@
 CLAIM_LOCK_BLOCKING_TIMEOUT_S = 5
 DEFAULT_SYSTEM_CONFIG_DIR = "/etc/fastloop.d/"
 LEASE_TTL_S = 30
-LEASE_HEARTBEAT_INTERVAL_S = 10
+LEASE_HEARTBEAT_INTERVAL_S = 5
 MAX_EVENT_HISTORY = 1000
 MEANINGFUL_WORK_THRESHOLD_S = 0.01

fastloop/context.py

@@ -60,6 +60,7 @@ def __init__(
         self.integration_events: dict[str, list[Any]] = {
             i.type(): i.events() for i in integrations
         }
+        self._integration_clients: dict[str, Any] = {}
 
     def _reset_cycle_tracking(self) -> None:
         self.event_this_cycle = False
@@ -199,7 +200,7 @@ async def _emit_to_integrations(self, event: LoopEvent) -> None:
                         },
                     )
 
-                    await self.integrations[integration_type].emit(event)
+                    await self.integrations[integration_type].emit(event, self)
 
     async def set(self, key: str, value: Any, local: bool = False) -> None:
         if not local:
@@ -231,6 +232,28 @@ async def delete(self, key: str, local: bool = False) -> None:
     async def get_event_history(self) -> list[dict[str, Any]]:
         return await self.state_manager.get_event_history(self.loop_id)
 
+    def set_integration_client(self, integration_type: str, client: Any) -> None:
+        self._integration_clients[integration_type] = client
+
+    def get_integration_client(self, integration_type: str) -> Any | None:
+        return self._integration_clients.get(integration_type)
+
+    async def setup_integrations(self, event: LoopEvent | None = None) -> None:
+        if not self.integrations:
+            return
+
+        event_to_use = event or self.initial_event
+        if not event_to_use:
+            logger.warning(
+                "Cannot setup integrations: no event available",
+                extra={"loop_id": self.loop_id},
+            )
+            return
+
+        for integration in self.integrations.values():
+            if hasattr(integration, "setup_for_context"):
+                await integration.setup_for_context(self, event_to_use)
+
     @property
     def should_stop(self) -> bool:
         """Check if the loop should stop."""

fastloop/executor.py

@@ -12,12 +12,15 @@
 
 def _get_pool(executor_type: ExecutorType) -> ThreadPoolExecutor | ProcessPoolExecutor:
     global _thread_pool, _process_pool
+
     if executor_type == ExecutorType.THREAD:
         if _thread_pool is None:
             _thread_pool = ThreadPoolExecutor(max_workers=4)
         return _thread_pool
+
     if _process_pool is None:
         _process_pool = ProcessPoolExecutor(max_workers=4)
+
     return _process_pool
 
 

fastloop/fastloop.py

@@ -351,6 +351,7 @@ async def _event_handler(request: dict[str, Any], func: Any = func):
                     integrations=self._loop_metadata[name].get("integrations", []),
                 )
 
+                await context.setup_integrations(event)
                 await self.state_manager.push_event(loop.loop_id, event)
 
                 if loop_instance:
@@ -697,6 +698,8 @@ async def restart_loop(self, loop_id: str) -> bool:
                 integrations=metadata.get("integrations", []),
             )
 
+            await context.setup_integrations()
+
             loop_instance: Loop | None = metadata.get("loop_instance")
             if loop_instance:
                 loop_instance.ctx = context
@@ -718,7 +721,7 @@ async def restart_loop(self, loop_id: str) -> bool:
                 return True
             else:
                 logger.warning(
-                    "Failed to restart loop",
+                    "Failed to restart loop - task already exists in loop_manager",
                     extra={
                         "loop_id": loop.loop_id,
                     },

fastloop/integrations/__init__.py

@@ -4,7 +4,8 @@
 from ..types import IntegrationType
 
 if TYPE_CHECKING:
-    from ..app import FastLoop
+    from ..context import LoopContext
+    from ..fastloop import FastLoop
 
 
 class Integration(ABC):
@@ -17,7 +18,7 @@ def register(self, fastloop: "FastLoop", loop_name: str) -> None:
         pass
 
     @abstractmethod
-    async def emit(self, event: Any) -> None:
+    async def emit(self, event: Any, context: "LoopContext | None" = None) -> None:
         pass
 
     @abstractmethod