add workflows and class based

Author: luke-beamcloud

examples/class_based.py

@@ -0,0 +1,57 @@
+"""
+Example of class-based loop definition.
+
+This demonstrates the class-based alternative to the function decorator approach.
+Both styles use the same @app.loop decorator and have identical functionality.
+"""
+
+from fastloop import FastLoop, Loop, LoopContext, LoopEvent
+
+app = FastLoop(name="class-based-demo")
+
+
+@app.event("chat.message")
+class ChatMessage(LoopEvent):
+    content: str
+
+
+@app.event("chat.response")
+class ChatResponse(LoopEvent):
+    content: str
+
+
+@app.loop("chat", start_event=ChatMessage)
+class ChatLoop(Loop):
+    """A chat loop implemented as a class."""
+
+    messages_processed: int = 0
+
+    async def on_start(self, ctx: LoopContext) -> None:
+        print(f"[{ctx.loop_id}] Chat loop started")
+        self.messages_processed = 0
+
+    async def on_stop(self, ctx: LoopContext) -> None:
+        print(
+            f"[{ctx.loop_id}] Chat loop stopped, processed {self.messages_processed} messages"
+        )
+
+    async def on_app_start(self, ctx: LoopContext) -> bool:
+        print(f"[{ctx.loop_id}] App started, checking if loop should resume...")
+        pending = await ctx.get("pending_work", default=False)
+        return pending
+
+    async def on_event(self, ctx: LoopContext, event: LoopEvent) -> None:
+        print(f"[{ctx.loop_id}] Received event: {event.type}")
+
+    async def loop(self, ctx: LoopContext) -> None:
+        message = await ctx.wait_for(ChatMessage, timeout=5, raise_on_timeout=False)
+
+        if message:
+            self.messages_processed += 1
+            response = f"Echo: {message.content}"
+            await ctx.emit(ChatResponse(content=response))
+            print(f"[{ctx.loop_id}] Processed message #{self.messages_processed}")
+
+
+if __name__ == "__main__":
+    app.run(port=8112)

examples/workflows.py

@@ -0,0 +1,114 @@
+"""
+Workflow Example - Durable multi-step execution.
+
+Control flow:
+  ctx.next(payload)  → advance to next block
+  ctx.repeat()       → re-run current block
+  normal return      → workflow completes
+"""
+
+from fastloop import FastLoop, LoopContext, LoopEvent, WorkflowBlock
+
+app = FastLoop(name="workflow-demo")
+
+
+# --- Events ---
+
+
+@app.event("start_workflow")
+class StartWorkflow(LoopEvent):
+    pass
+
+
+@app.event("user_input")
+class UserInput(LoopEvent):
+    value: str
+
+
+# --- Callbacks (optional) ---
+
+
+def on_block_done(ctx: LoopContext, block: WorkflowBlock, payload: dict | None):
+    print(f"  ✓ Block complete: {block.type}")
+
+
+def on_error(ctx: LoopContext, block: WorkflowBlock, error: Exception):
+    print(f"  ✗ Block error: {block.type} - {error}")
+
+
+# --- Workflow ---
+
+
+@app.workflow(
+    name="onboarding",
+    start_event=StartWorkflow,
+    on_block_complete=on_block_done,
+    on_error=on_error,
+)
+async def onboarding_workflow(
+    ctx: LoopContext,
+    blocks: list[WorkflowBlock],
+    current_block: WorkflowBlock,
+):
+    """
+    Multi-step onboarding. State survives restarts.
+
+    ctx.block_index      - current position (0-indexed)
+    ctx.block_count      - total blocks
+    ctx.previous_payload - data from previous ctx.next(payload)
+    """
+    print(f"[{ctx.block_index + 1}/{ctx.block_count}] {current_block.type}")
+
+    match current_block.type:
+        case "collect_name":
+            response = await ctx.wait_for(
+                UserInput, timeout=300.0, raise_on_timeout=False
+            )
+            if response:
+                await ctx.set("user_name", response.value)
+                ctx.next()
+            else:
+                ctx.repeat()
+
+        case "collect_email":
+            response = await ctx.wait_for(
+                UserInput, timeout=300.0, raise_on_timeout=False
+            )
+            if response:
+                await ctx.set("user_email", response.value)
+                ctx.next()
+            else:
+                ctx.repeat()
+
+        case "confirm":
+            name = await ctx.get("user_name")
+            email = await ctx.get("user_email")
+            print(f"Done: {name} <{email}>")
+
+
+if __name__ == "__main__":
+    app.run(port=8111)
+
+# Example usage:
+#
+# 1. Start workflow:
+#    curl -X POST http://localhost:8111/onboarding \
+#      -H "Content-Type: application/json" \
+#      -d '{
+#        "type": "start_workflow",
+#        "blocks": [
+#          {"type": "collect_name", "text": "Please enter your name"},
+#          {"type": "collect_email", "text": "Please enter your email"},
+#          {"type": "confirm", "text": "Confirm your details"}
+#        ]
+#      }'
+#
+# 2. Send event to workflow (use workflow_id from response):
+#    curl -X POST http://localhost:8111/onboarding/<workflow_id>/event \
+#      -H "Content-Type: application/json" \
+#      -d '{"type": "user_input", "value": "John Doe", "workflow_id": "<id>"}'
+#
+# 3. Check workflow status:
+#    curl http://localhost:8111/onboarding/<workflow_id>
+#
+# 4. If service restarts, workflow resumes from current block automatically

fastloop/__init__.py

@@ -1,11 +1,13 @@
 from . import integrations
 from .context import LoopContext
 from .fastloop import FastLoop
-from .loop import LoopEvent
+from .loop import Loop, LoopEvent, WorkflowBlock
 
 __all__ = [
     "FastLoop",
+    "Loop",
     "LoopContext",
     "LoopEvent",
+    "WorkflowBlock",
     "integrations",
 ]

fastloop/context.py

@@ -10,6 +10,8 @@
     LoopContextSwitchError,
     LoopPausedError,
     LoopStoppedError,
+    WorkflowNextError,
+    WorkflowRepeatError,
 )
 from .logging import setup_logger
 from .loop import LoopEvent
@@ -51,19 +53,12 @@ def __init__(
         self.state_manager: StateManager = state_manager
         self.event_this_cycle: bool = False
 
-        if integrations is None:
-            integrations = []
-
-        self.integrations: dict[str, Integration] = {
-            integration.type(): integration for integration in integrations
+        integrations = integrations or []
+        self.integrations: dict[str, Integration] = {i.type(): i for i in integrations}
+        self.integration_events: dict[str, list[Any]] = {
+            i.type(): i.events() for i in integrations
         }
 
-        self.integration_events: dict[str, list[Any]] = {}
-
-        for integration in integrations:
-            self.integrations[integration.type()] = integration
-            self.integration_events[integration.type()] = integration.events()
-
     def stop(self):
         """Request the loop to stop on the next iteration."""
         self._stop_requested = True
@@ -250,3 +245,11 @@ def _parse_duration(self, duration_str: str) -> float:
             if unit.startswith(prefix):
                 return value * mult
         return value
+
+    def next(self, payload: dict | None = None) -> None:
+        """Advance to the next block in a workflow."""
+        raise WorkflowNextError(payload)
+
+    def repeat(self) -> None:
+        """Re-execute the current block in a workflow."""
+        raise WorkflowRepeatError()

fastloop/exceptions.py

@@ -39,3 +39,16 @@ class LoopContextSwitchError(Exception):
     def __init__(self, func: Callable[[T], Awaitable[None]], context: "LoopContext"):
         self.func = func
         self.context = context
+
+
+class WorkflowNextError(Exception):
+    def __init__(self, payload: dict | None = None):
+        self.payload = payload
+
+
+class WorkflowRepeatError(Exception):
+    pass
+
+
+class WorkflowNotFoundError(Exception):
+    pass