Merge pull request #4 from flamingo-run/feature/subworkflows

Subworkflows

Author: joaodaher

README.md

@@ -116,7 +116,7 @@ Priority order for upcoming features:
 | --- | --- | --- | --- |
 | 1 | **Retries** | ✅ Completed | `RetryPolicy` with configurable backoff and predicates |
 | 2 | **Try/catch** | ✅ Completed | `TryCatchStep` for exception handling with fallback flows |
-| 3 | **Subworkflows** | 📋 Planned | Call other workflows, composition patterns |
+| 3 | **Subworkflows** | ✅ Completed | Call other workflows with `>> WORKFLOW`, dependency tracking |
 | 4 | **GCP connectors** | 📋 Planned | Direct service calls, native Cloud Workflows connectors |
 | 5 | **Deployment API** | 📋 Planned | Programmatic deployment of workflows & EventArc triggers via GCP APIs |
 | 6 | **Loops** | 📋 Planned | For/while constructs, iteration over collections |

examples/app/flows/order_with_subworkflow.py

@@ -0,0 +1,84 @@
+"""Order processing with payment as a subworkflow."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from fastapi_cloudflow import Context, step, workflow
+
+
+# Payment workflow models and steps
+class PaymentRequest(BaseModel):
+    amount: float
+    customer_id: str
+    payment_method: str
+
+
+class PaymentResult(BaseModel):
+    transaction_id: str
+    status: str
+    amount: float
+
+
+@step(name="validate-payment")
+async def validate_payment(ctx: Context, data: PaymentRequest) -> PaymentRequest:
+    """Validate payment details."""
+    if data.amount <= 0:
+        raise ValueError("Invalid payment amount")
+    return data
+
+
+@step(name="charge-card")
+async def charge_card(ctx: Context, data: PaymentRequest) -> PaymentResult:
+    """Process the payment."""
+    return PaymentResult(transaction_id=f"txn_{data.customer_id}_{data.amount}", status="success", amount=data.amount)
+
+
+# Build the payment workflow
+PAYMENT_WORKFLOW = (workflow("payment-processing") >> validate_payment >> charge_card).build()
+
+
+# Order workflow models and steps
+class OrderRequest(BaseModel):
+    order_id: str
+    customer_id: str
+    items: list[str]
+    total_amount: float
+
+
+class OrderValidated(BaseModel):
+    order_id: str
+    customer_id: str
+    amount: float
+    payment_method: str
+
+
+class OrderComplete(BaseModel):
+    order_id: str
+    transaction_id: str
+    status: str
+
+
+@step(name="validate-order")
+async def validate_order(ctx: Context, data: OrderRequest) -> PaymentRequest:
+    """Validate the order and prepare payment request."""
+    return PaymentRequest(amount=data.total_amount, customer_id=data.customer_id, payment_method="credit_card")
+
+
+@step(name="ship-order")
+async def ship_order(ctx: Context, data: PaymentResult) -> OrderComplete:
+    """Ship the order after successful payment."""
+    return OrderComplete(
+        order_id="order_123",  # Would normally come from context
+        transaction_id=data.transaction_id,
+        status="shipped",
+    )
+
+
+# Build the order workflow using payment as a subworkflow
+ORDER_WORKFLOW = (
+    workflow("order-processing")
+    >> validate_order
+    >> PAYMENT_WORKFLOW  # Use payment workflow as a subworkflow
+    >> ship_order
+).build()

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastapi-cloudflow"
-version = "0.0.2"
+version = "0.0.3"
 description = "Typed Python-to-Google Cloud Workflows framework with FastAPI codegen"
 readme = "README.md"
 requires-python = ">=3.13"

src/fastapi_cloudflow/__init__.py

@@ -6,10 +6,12 @@
     ModelAdapter,
     RetryPolicy,
     Step,
+    SubworkflowStep,
     TryCatchBuilder,
     TryCatchStep,
     Workflow,
     get_registry,
+    get_workflow_dependencies,
     get_workflows,
     step,
     try_catch,
@@ -25,13 +27,15 @@
     "Step",
     "Workflow",
     "RetryPolicy",
+    "SubworkflowStep",
     "TryCatchStep",
     "TryCatchBuilder",
     "AssignStep",
     "HttpStep",
     "ModelAdapter",
     "Arg",
     "get_registry",
+    "get_workflow_dependencies",
     "get_workflows",
     "attach_to_fastapi",
     "build_app",

src/fastapi_cloudflow/codegen/workflows.py

@@ -80,6 +80,35 @@ def _process_single_step(
         )
         return steps, have_run_id
 
+    # Handle SubworkflowStep
+    from fastapi_cloudflow.core.subworkflow import SubworkflowStep
+
+    if isinstance(node, SubworkflowStep):
+        result_var = f"res_{idx}"
+        args = {"workflow_id": node.workflow_id, "argument": f"${{{payload_var}}}"}
+
+        # Apply input mapping if specified
+        if node.input_mapping:
+            mapped_args = {k: _as_yaml_expr(v) for k, v in node.input_mapping.items()}
+            args["argument"] = f"${mapped_args}"
+
+        step_def: dict[str, Any] = {"call": "workflows.executeWorkflow", "args": args}
+
+        if node.wait:
+            step_def["result"] = result_var
+            steps.append({f"call_{node.name}": step_def})
+            # Extract result and apply output mapping
+            if node.output_mapping:
+                mapped_output = {k: f"${{{result_var}.{v}}}" for k, v in node.output_mapping.items()}
+                steps.append({f"map_output_{idx}": {"assign": [{payload_var: mapped_output}]}})
+            else:
+                steps.append({f"set_payload_{idx}": {"assign": [{payload_var: f"${{{result_var}}}"}]}})
+        else:
+            # Fire-and-forget mode (no result)
+            steps.append({f"call_{node.name}": step_def})
+
+        return steps, have_run_id
+
     if isinstance(node, HttpStep):
         method = node.method.lower()
         result_var = f"res_{idx}"

src/fastapi_cloudflow/core/__init__.py

@@ -1,12 +1,14 @@
 from fastapi_cloudflow.core.arg import Arg, ArgExpr
 from fastapi_cloudflow.core.error_handling import TryCatchBuilder, TryCatchStep, try_catch
 from fastapi_cloudflow.core.step import AssignStep, HttpStep, ModelAdapter, Step
+from fastapi_cloudflow.core.subworkflow import SubworkflowStep
 from fastapi_cloudflow.core.types import Context, RetryPolicy, WorkflowMeta
 from fastapi_cloudflow.core.workflow import (
     Registry,
     Workflow,
     WorkflowBuilder,
     get_registry,
+    get_workflow_dependencies,
     get_workflows,
     step,
     workflow,
@@ -22,6 +24,7 @@
     "AssignStep",
     "HttpStep",
     "ModelAdapter",
+    "SubworkflowStep",
     "TryCatchStep",
     "TryCatchBuilder",
     "try_catch",
@@ -30,6 +33,7 @@
     "WorkflowBuilder",
     "workflow",
     "get_registry",
+    "get_workflow_dependencies",
     "get_workflows",
     "step",
 ]

src/fastapi_cloudflow/core/subworkflow.py

@@ -0,0 +1,52 @@
+"""Subworkflow support for Cloud Workflows."""
+
+from __future__ import annotations
+
+from typing import Any, TypeVar
+
+from pydantic import BaseModel
+
+from fastapi_cloudflow.core.step import Step
+
+InT = TypeVar("InT", bound=BaseModel)
+OutT = TypeVar("OutT", bound=BaseModel)
+
+
+class SubworkflowStep[InT: BaseModel, OutT: BaseModel](Step[InT, OutT]):
+    """
+    A step that executes another workflow.
+
+    This step represents a call to another Cloud Workflow, enabling
+    workflow composition and reusability.
+    """
+
+    def __init__(
+        self,
+        workflow_id: str,
+        input_model: type[InT],
+        output_model: type[OutT],
+        wait: bool = True,
+        input_mapping: dict[str, Any] | None = None,
+        output_mapping: dict[str, Any] | None = None,
+    ) -> None:
+        """
+        Initialize a subworkflow step.
+
+        Args:
+            workflow_id: The ID of the workflow to call
+            input_model: The input model type for this step
+            output_model: The output model type for this step
+            wait: Whether to wait for the subworkflow to complete (default: True)
+            input_mapping: Optional mapping for input parameters
+            output_mapping: Optional mapping for output results
+        """
+        super().__init__(
+            name=f"call_{workflow_id.replace('-', '_')}",
+            input_model=input_model,
+            output_model=output_model,
+            fn=None,  # Subworkflows are not callable locally
+        )
+        self.workflow_id = workflow_id
+        self.wait = wait
+        self.input_mapping = input_mapping or {}
+        self.output_mapping = output_mapping or {}

src/fastapi_cloudflow/core/workflow.py

@@ -14,6 +14,24 @@ class Workflow:
     def __init__(self, name: str, nodes: list[Step[Any, Any]]) -> None:
         self.name = name
         self.nodes = nodes
+        self.dependencies: set[str] = set()
+
+    def to_subworkflow_step(self) -> Step[Any, Any]:
+        """Convert this workflow to a SubworkflowStep for composition."""
+        from fastapi_cloudflow.core.subworkflow import SubworkflowStep
+
+        # Get input/output models from first and last steps
+        if not self.nodes:
+            raise ValueError(f"Workflow '{self.name}' has no steps")
+
+        first_step = self.nodes[0]
+        last_step = self.nodes[-1]
+
+        return SubworkflowStep(
+            workflow_id=self.name,
+            input_model=first_step.input_model,
+            output_model=last_step.output_model,
+        )
 
 
 class Registry:
@@ -44,21 +62,36 @@ class WorkflowBuilder:
     def __init__(self, name: str, nodes: list[Step[Any, Any]] | None = None) -> None:
         self.name = name
         self.nodes = nodes or []
+        self._dependencies: set[str] = set()
+
+    def __rshift__(self, other: Step[Any, Any] | Workflow) -> WorkflowBuilder:
+        # Convert Workflow to SubworkflowStep if needed
+        if isinstance(other, Workflow):
+            step = other.to_subworkflow_step()
+            # Track dependency
+            self._dependencies.add(other.name)
+        else:
+            step = other
 
-    def __rshift__(self, other: Step[Any, Any]) -> WorkflowBuilder:
         if self.nodes:
             prev = self.nodes[-1]
-            if prev.output_model is not other.input_model:
+            if prev.output_model is not step.input_model:
                 raise TypeError(
                     f"Type mismatch: {prev.name} outputs {prev.output_model.__name__} "
-                    f"but {other.name} expects {other.input_model.__name__}"
+                    f"but {step.name} expects {step.input_model.__name__}"
                 )
-        return WorkflowBuilder(self.name, self.nodes + [other])
+
+        # Create new builder with updated nodes and preserve dependencies
+        new_builder = WorkflowBuilder(self.name, self.nodes + [step])
+        new_builder._dependencies = self._dependencies.copy()
+        return new_builder
 
     def build(self) -> Workflow:
         if not self.nodes:
             raise ValueError("Workflow has no steps")
         wf = Workflow(self.name, self.nodes)
+        # Transfer any tracked dependencies
+        wf.dependencies = self._dependencies
         _REGISTRY.register_workflow(wf)
         return wf
 
@@ -109,3 +142,12 @@ def get_registry() -> Registry:
 
 def get_workflows() -> list[Workflow]:
     return _REGISTRY.get_workflows()
+
+
+def get_workflow_dependencies() -> dict[str, set[str]]:
+    """Get all workflow dependencies as a mapping."""
+    deps: dict[str, set[str]] = {}
+    for wf in _REGISTRY.workflows.values():
+        if wf.dependencies:
+            deps[wf.name] = wf.dependencies
+    return deps