wip

Author: luke-beamcloud

fastloop/fastloop.py

@@ -38,7 +38,7 @@
 from .models import LoopEvent
 from .monitor import LoopMonitor
 from .scheduler import Schedule, validate_cron
-from .state.state import StateManager, create_state_manager
+from .state.state import LoopState, StateManager, create_state_manager
 from .task import TaskManager, TaskResult
 from .types import BaseConfig, ExecutorType, LoopStatus, RetryPolicy
 from .utils import get_func_import_path, import_func_from_path, infer_application_path
@@ -682,7 +682,7 @@ async def _cancel_handler(workflow_run_id: str):
                     ) from e
 
             async def _resume_handler(
-                workflow_run_id: str, request: dict[str, Any] = {}
+                workflow_run_id: str, request: dict[str, Any] | None = None
             ):
                 try:
                     workflow = await self.state_manager.get_workflow(workflow_run_id)
@@ -829,6 +829,166 @@ async def has_active_clients(self, loop_id: str) -> bool:
         client_count = await self.state_manager.get_active_client_count(loop_id)
         return client_count > 0
 
+    async def start_loop(
+        self,
+        name: str,
+        loop_id: str,
+        initial_data: dict[str, Any] | None = None,
+    ) -> LoopState:
+        """Start a named loop with a specific loop_id.
+
+        Args:
+            name: The registered loop name (from @app.loop decorator)
+            loop_id: The unique identifier for this loop instance
+            initial_data: Optional initial context data for the loop
+
+        Returns:
+            The LoopState for the started loop
+
+        Raises:
+            LoopNotFoundError: If the loop name is not registered
+        """
+        if name not in self._loop_metadata:
+            raise LoopNotFoundError(f"Loop '{name}' is not registered")
+
+        metadata = self._loop_metadata[name]
+        func = metadata["func"]
+
+        loop, _created = await self.state_manager.get_or_create_loop(
+            loop_name=name,
+            loop_id=loop_id,
+            current_function_path=get_func_import_path(func),
+            create_with_id=True,
+        )
+
+        if loop.status == LoopStatus.STOPPED:
+            logger.warning(
+                "Loop is stopped, not starting",
+                extra={"loop_id": loop_id, "loop_name": name},
+            )
+            return loop
+
+        context = LoopContext(
+            loop_id=loop.loop_id,
+            initial_event=None,
+            state_manager=self.state_manager,
+            integrations=metadata.get("integrations", []),
+        )
+
+        if initial_data:
+            for key, value in initial_data.items():
+                await context.set(key, value)
+
+        await context.setup_integrations()
+
+        loop_instance: Loop | None = metadata.get("loop_instance")
+        if loop_instance:
+            loop_instance.ctx = context
+            func = loop_instance.loop
+
+        started = await self.loop_manager.start(
+            func=func,
+            loop_start_func=metadata.get("on_start"),
+            loop_stop_func=metadata.get("on_stop"),
+            context=context,
+            loop=loop,
+            loop_delay=metadata["loop_delay"],
+            stop_after_idle_seconds=metadata.get("stop_after_idle_seconds"),
+            pause_after_idle_seconds=metadata.get("pause_after_idle_seconds"),
+        )
+
+        if started:
+            logger.info(
+                "Loop started",
+                extra={"loop_id": loop.loop_id, "loop_name": name},
+            )
+
+        return await self.state_manager.get_loop(loop.loop_id)
+
+    async def stop_loop(self, name: str, loop_id: str) -> bool:
+        """Stop a specific loop instance.
+
+        Args:
+            name: The registered loop name
+            loop_id: The unique identifier for this loop instance
+
+        Returns:
+            True if the loop was stopped, False if it wasn't running
+        """
+        if name not in self._loop_metadata:
+            raise LoopNotFoundError(f"Loop '{name}' is not registered")
+
+        try:
+            loop = await self.state_manager.get_loop(loop_id)
+            if loop.loop_name != name:
+                logger.warning(
+                    "Loop name mismatch",
+                    extra={
+                        "expected": name,
+                        "actual": loop.loop_name,
+                        "loop_id": loop_id,
+                    },
+                )
+                return False
+
+            await self.state_manager.update_loop_status(loop_id, LoopStatus.STOPPED)
+            stopped = await self.loop_manager.stop(loop_id)
+
+            if stopped:
+                logger.info(
+                    "Loop stopped",
+                    extra={"loop_id": loop_id, "loop_name": name},
+                )
+
+            return stopped
+
+        except LoopNotFoundError:
+            return False
+
+    async def loop_exists(self, name: str, loop_id: str) -> bool:
+        """Check if a loop instance exists and is active (running or idle).
+
+        Args:
+            name: The registered loop name
+            loop_id: The unique identifier for this loop instance
+
+        Returns:
+            True if the loop exists and is running/idle, False otherwise
+        """
+        if name not in self._loop_metadata:
+            return False
+
+        try:
+            loop = await self.state_manager.get_loop(loop_id)
+            if loop.loop_name != name:
+                return False
+            return loop.status in (
+                LoopStatus.RUNNING,
+                LoopStatus.IDLE,
+                LoopStatus.PENDING,
+            )
+        except LoopNotFoundError:
+            return False
+
+    async def list_loops(self, name: str) -> list[str]:
+        """List all active loop IDs for a given loop name.
+
+        Args:
+            name: The registered loop name
+
+        Returns:
+            List of loop_ids that are currently active (running or idle)
+        """
+        if name not in self._loop_metadata:
+            raise LoopNotFoundError(f"Loop '{name}' is not registered")
+
+        loops = await self.state_manager.get_loops_by_name(name)
+        return [
+            loop.loop_id
+            for loop in loops
+            if loop.status in (LoopStatus.RUNNING, LoopStatus.IDLE, LoopStatus.PENDING)
+        ]
+
     async def restart_workflow(self, workflow_run_id: str) -> bool:
         """Restart a workflow from its persisted state."""
         try:

fastloop/state/state.py

@@ -46,6 +46,7 @@ async def get_or_create_loop(
         loop_name: str | None = None,
         loop_id: str | None = None,
         current_function_path: str = "",
+        create_with_id: bool = False,
     ) -> tuple[LoopState, bool]:
         pass
 
@@ -57,6 +58,16 @@ async def update_loop(self, loop_id: str, state: LoopState):
     async def update_loop_status(self, loop_id: str, status: LoopStatus) -> LoopState:
         pass
 
+    @abstractmethod
+    async def get_loops_by_name(
+        self, loop_name: str, status: LoopStatus | None = None
+    ) -> list[LoopState]:
+        pass
+
+    @abstractmethod
+    async def add_loop_to_name_index(self, loop_name: str, loop_id: str) -> None:
+        pass
+
     @abstractmethod
     async def get_event_history(self, loop_id: str) -> list[dict[str, Any]]:
         pass

fastloop/state/state_redis.py

@@ -40,6 +40,7 @@
 
 class RedisKeys:
     LOOP_INDEX = f"{KEY_PREFIX}:{{app_name}}:index"
+    LOOP_NAME_INDEX = f"{KEY_PREFIX}:{{app_name}}:loops_by_name:{{loop_name}}"
     LOOP_EVENT_QUEUE_SERVER = f"{KEY_PREFIX}:{{app_name}}:events:{{loop_id}}:server"
     LOOP_EVENT_QUEUE_CLIENT = (
         f"{KEY_PREFIX}:{{app_name}}:events:{{loop_id}}:{{event_type}}:client"
@@ -339,20 +340,23 @@ async def get_or_create_loop(
         loop_name: str | None = None,
         loop_id: str | None = None,
         current_function_path: str = "",
+        create_with_id: bool = False,
     ) -> tuple[LoopState, bool]:
         if loop_id:
             loop_str = await self.rdb.get(
                 RedisKeys.LOOP_STATE.format(app_name=self.app_name, loop_id=loop_id)
             )
             if loop_str:
                 return LoopState.from_json(loop_str.decode("utf-8")), False
-            else:
+            elif not create_with_id:
                 raise LoopNotFoundError(f"Loop {loop_id} not found")
 
         if not current_function_path:
             raise ValueError("Current function is required")
 
-        loop_id = str(uuid.uuid4())
+        if not loop_id:
+            loop_id = str(uuid.uuid4())
+
         loop = LoopState(
             loop_id=loop_id,
             loop_name=loop_name,
@@ -368,6 +372,9 @@ async def get_or_create_loop(
             RedisKeys.LOOP_INDEX.format(app_name=self.app_name), loop_id
         )  # type: ignore
 
+        if loop_name:
+            await self.add_loop_to_name_index(loop_name, loop_id)
+
         return loop, True
 
     async def update_loop(self, loop_id: str, state: LoopState):
@@ -554,6 +561,50 @@ async def get_all_loops(self, status: LoopStatus | None = None) -> list[LoopStat
 
         return results
 
+    async def get_loops_by_name(
+        self, loop_name: str, status: LoopStatus | None = None
+    ) -> list[LoopState]:
+        name_index_key = RedisKeys.LOOP_NAME_INDEX.format(
+            app_name=self.app_name, loop_name=loop_name
+        )
+        loop_ids = await self.rdb.smembers(name_index_key)
+        if not loop_ids:
+            return []
+
+        keys = [
+            RedisKeys.LOOP_STATE.format(app_name=self.app_name, loop_id=lid.decode())
+            for lid in loop_ids
+        ]
+        values = await self.rdb.mget(keys)
+
+        results: list[LoopState] = []
+        stale_ids: list[str] = []
+
+        for loop_id_bytes, val in zip(loop_ids, values, strict=True):
+            loop_id = loop_id_bytes.decode()
+            if not val:
+                stale_ids.append(loop_id)
+                continue
+            try:
+                loop_state = LoopState.from_json(val.decode("utf-8"))
+            except (TypeError, json.JSONDecodeError):
+                stale_ids.append(loop_id)
+                continue
+            if status and loop_state.status != status:
+                continue
+            results.append(loop_state)
+
+        if stale_ids:
+            await self.rdb.srem(name_index_key, *stale_ids)
+
+        return results
+
+    async def add_loop_to_name_index(self, loop_name: str, loop_id: str) -> None:
+        name_index_key = RedisKeys.LOOP_NAME_INDEX.format(
+            app_name=self.app_name, loop_name=loop_name
+        )
+        await self.rdb.sadd(name_index_key, loop_id)
+
     async def get_event_history(self, loop_id: str) -> list[dict[str, Any]]:
         event_history: list[bytes] | None = await self.rdb.lrange(  # type: ignore
             RedisKeys.LOOP_EVENT_HISTORY.format(

fastloop/state/state_s3.py

@@ -21,6 +21,10 @@ class S3Keys:
     def loop_index(prefix: str, app_name: str) -> str:
         return f"{prefix}/{app_name}/index.json"
 
+    @staticmethod
+    def loop_name_index(prefix: str, app_name: str, loop_name: str) -> str:
+        return f"{prefix}/{app_name}/loops_by_name/{loop_name}.json"
+
     @staticmethod
     def loop_state(prefix: str, app_name: str, loop_id: str) -> str:
         return f"{prefix}/{app_name}/state/{loop_id}.json"
@@ -264,12 +268,18 @@ async def get_or_create_loop(
         loop_name: str | None = None,
         loop_id: str | None = None,
         current_function_path: str = "",
+        create_with_id: bool = False,
     ) -> tuple[LoopState, bool]:
         if loop_id:
-            loop = await self.get_loop(loop_id)
-            return loop, False
+            try:
+                loop = await self.get_loop(loop_id)
+                return loop, False
+            except LoopNotFoundError:
+                if not create_with_id:
+                    raise
 
-        loop_id = str(uuid.uuid4())
+        if not loop_id:
+            loop_id = str(uuid.uuid4())
         loop = LoopState(
             loop_id=loop_id,
             loop_name=loop_name,
@@ -285,6 +295,9 @@ async def get_or_create_loop(
         index.append(loop_id)
         self._put_json(S3Keys.loop_index(self.prefix, self.app_name), index)
 
+        if loop_name:
+            await self.add_loop_to_name_index(loop_name, loop_id)
+
         return loop, True
 
     async def update_loop(self, loop_id: str, state: LoopState):
@@ -316,6 +329,31 @@ async def get_all_loops(self, status: LoopStatus | None = None) -> list[LoopStat
 
         return loops
 
+    async def get_loops_by_name(
+        self, loop_name: str, status: LoopStatus | None = None
+    ) -> list[LoopState]:
+        name_index_key = S3Keys.loop_name_index(self.prefix, self.app_name, loop_name)
+        loop_ids: list[str] = self._get_json(name_index_key) or []
+        loops: list[LoopState] = []
+
+        for loop_id in loop_ids:
+            try:
+                loop = await self.get_loop(loop_id)
+                if status and loop.status != status:
+                    continue
+                loops.append(loop)
+            except LoopNotFoundError:
+                continue
+
+        return loops
+
+    async def add_loop_to_name_index(self, loop_name: str, loop_id: str) -> None:
+        name_index_key = S3Keys.loop_name_index(self.prefix, self.app_name, loop_name)
+        loop_ids: list[str] = self._get_json(name_index_key) or []
+        if loop_id not in loop_ids:
+            loop_ids.append(loop_id)
+            self._put_json(name_index_key, loop_ids)
+
     async def get_event_history(self, loop_id: str) -> list[dict[str, Any]]:
         return (
             self._get_json(
@@ -749,10 +787,8 @@ async def set_workflow_resume_payload(
             f"{self.prefix}/{self.app_name}/workflow_resume_payload/{workflow_id}.json"
         )
         if payload is None:
-            try:
+            with suppress(ClientError):
                 self.s3.delete_object(Bucket=self.bucket, Key=key)
-            except ClientError:
-                pass
         else:
             self.s3.put_object(
                 Bucket=self.bucket, Key=key, Body=json.dumps(payload).encode("utf-8")

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastloop"
-version = "0.1.116"
+version = "0.1.117"
 description = "A Python package for deploying stateful loops"
 readme = "README.md"
 requires-python = ">=3.12"