Merge pull request #5 from flamingo-run/feature/pubsub

Enables Pub/Sub connector for workflows

Author: joaodaher

README.md

@@ -117,7 +117,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** | ✅ Completed | Call other workflows with `>> WORKFLOW`, dependency tracking |
-| 4 | **GCP connectors** | 📋 Planned | Direct service calls, native Cloud Workflows connectors |
+| 4 | **GCP connectors** | ✅ Completed | Direct service calls using Cloud Workflows connectors (Pub/Sub first) |
 | 5 | **Deployment API** | 📋 Planned | Programmatic deployment of workflows & EventArc triggers via GCP APIs |
 | 6 | **Loops** | 📋 Planned | For/while constructs, iteration over collections |
 | 7 | **Conditionals / switch** | 📋 Planned | Branching logic, switch statements |

conftest.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT_DIR = Path(__file__).resolve().parent
+PROJECT_ROOT = ROOT_DIR
+EXAMPLES_DIR = PROJECT_ROOT / "examples"
+SRC_DIR = PROJECT_ROOT / "src"
+
+for path in [SRC_DIR, EXAMPLES_DIR, PROJECT_ROOT]:
+    path_str = str(path)
+    if path_str not in sys.path:
+        sys.path.insert(0, path_str)

examples/__init__.py

@@ -0,0 +1 @@
+"""Example application package for fastapi-cloudflow demos."""

examples/app/__init__.py

@@ -0,0 +1,12 @@
+from .flows.resilient_payment import GATEWAY_PAYMENT_FLOW, RESILIENT_PAYMENT_FLOW
+from .flows.retry_contract import RETRY_DEMO_WORKFLOW
+from .flows.subworkflow_contract import SUBWORKFLOW_DEMO
+from .flows.try_catch_contract import TRY_CATCH_DEMO
+
+__all__ = [
+    "GATEWAY_PAYMENT_FLOW",
+    "RESILIENT_PAYMENT_FLOW",
+    "RETRY_DEMO_WORKFLOW",
+    "SUBWORKFLOW_DEMO",
+    "TRY_CATCH_DEMO",
+]

examples/app/flows/__init__.py

@@ -0,0 +1,17 @@
+"""Example flow definitions for FastAPI Cloudflow demos."""
+
+__all__ = [
+    "data_pipeline",
+    "echo_name",
+    "jokes",
+    "order",
+    "order_with_subworkflow",
+    "payments",
+    "post_story",
+    "pubsub_example",
+    "resilient_payment",
+    "retry_contract",
+    "subworkflow_contract",
+    "try_catch_contract",
+    "user",
+]

examples/app/flows/pubsub_example.py

@@ -0,0 +1,33 @@
+"""Example workflow demonstrating Pub/Sub connector support."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from fastapi_cloudflow import Context, step, workflow
+from fastapi_cloudflow.core.connectors import pubsub_publish_step
+
+
+class PublishRequest(BaseModel):
+    topic: str
+    payload: str
+
+
+class PublishResult(BaseModel):
+    message_ids: list[str]
+
+
+@step(name="prepare-message")
+async def prepare_message(ctx: Context, data: PublishRequest) -> PublishRequest:
+    return data
+
+
+publish_step = pubsub_publish_step(
+    name="publish-topic",
+    topic="projects/demo/topics/example",
+    input_model=PublishRequest,
+    data_field="payload",
+)
+
+
+PUBSUB_EXAMPLE_FLOW = (workflow("pubsub-example-flow") >> prepare_message >> publish_step).build()

examples/app/flows/retry_contract.py

@@ -0,0 +1,57 @@
+"""Retry-focused workflow example used for documentation and contract tests."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+from pydantic import BaseModel
+
+from fastapi_cloudflow import Context, HttpStep, RetryPolicy, step, workflow
+
+
+class RetryInput(BaseModel):
+    value: int
+
+
+class RetryResult(BaseModel):
+    result: int
+
+
+@step(
+    name="retry-entry",
+    retry=RetryPolicy(
+        max_retries=3,
+        initial_delay_s=2.0,
+        max_delay_s=10.0,
+        multiplier=2.0,
+        predicate="http.default_retry_predicate",
+    ),
+    timeout=timedelta(seconds=30),
+)
+async def retry_entry(ctx: Context, data: RetryInput) -> RetryInput:
+    return RetryInput(value=data.value + 1)
+
+
+PAYMENT_GATEWAY = HttpStep(
+    name="retry-gateway",
+    input_model=RetryInput,
+    output_model=RetryResult,
+    method="POST",
+    url="https://api.example.com/payments",
+    retry=RetryPolicy.idempotent_http(),
+    timeout=timedelta(seconds=60),
+)
+
+
+@step(name="retry-finalize")
+async def retry_finalize(ctx: Context, data: RetryResult) -> RetryResult:
+    return RetryResult(result=data.result)
+
+
+def build_retry_demo_workflow():
+    """Create a workflow showcasing HTTP and step-level retry policies."""
+
+    return (workflow("retry-demo") >> retry_entry >> PAYMENT_GATEWAY >> retry_finalize).build()
+
+
+RETRY_DEMO_WORKFLOW = build_retry_demo_workflow()

examples/app/flows/subworkflow_contract.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from fastapi_cloudflow import Context, SubworkflowStep, step, workflow
+
+
+class SubworkflowInput(BaseModel):
+    value: int
+
+
+class SubworkflowResult(BaseModel):
+    result: int
+
+
+@step(name="subwf-prepare")
+async def subwf_prepare(ctx: Context, data: SubworkflowInput) -> SubworkflowInput:
+    return SubworkflowInput(value=data.value + 2)
+
+
+SUB_WORKFLOW = SubworkflowStep(
+    workflow_id="subwf-child",
+    input_model=SubworkflowInput,
+    output_model=SubworkflowResult,
+    input_mapping={"custom_field": "${payload.value}"},
+    output_mapping={"result": "custom_result"},
+)
+
+
+@step(name="subwf-finalize")
+async def subwf_finalize(ctx: Context, data: SubworkflowResult) -> SubworkflowResult:
+    return SubworkflowResult(result=data.result + 5)
+
+
+SUBWORKFLOW_DEMO = (workflow("subworkflow-demo") >> subwf_prepare >> SUB_WORKFLOW >> subwf_finalize).build()

examples/app/flows/try_catch_contract.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from fastapi_cloudflow import Context, step, try_catch, workflow
+
+
+class TryCatchInput(BaseModel):
+    value: int
+
+
+class TryCatchResult(BaseModel):
+    result: int
+
+
+@step(name="tc-risky")
+async def tc_risky(ctx: Context, data: TryCatchInput) -> TryCatchResult:
+    return TryCatchResult(result=data.value * 2)
+
+
+@step(name="tc-handler")
+async def tc_handler(ctx: Context, data: TryCatchInput) -> TryCatchResult:
+    return TryCatchResult(result=0)
+
+
+TRY_BLOCK = workflow("tc-try") >> tc_risky
+EXCEPT_BLOCK = workflow("tc-except") >> tc_handler
+TRY_CATCH_STEP = try_catch("tc-safe").try_block(TRY_BLOCK).except_block(EXCEPT_BLOCK).raise_error(True).build()
+
+
+TRY_CATCH_DEMO = (workflow("try-catch-demo") >> TRY_CATCH_STEP).build()

examples/app/main.py

@@ -1,14 +1,23 @@
 from fastapi import FastAPI
-from flows import data_pipeline, echo_name, jokes, order, payments, post_story, resilient_payment, user  # noqa: F401
 
+from examples.app.flows import (  # noqa: F401
+    data_pipeline,
+    echo_name,
+    jokes,
+    order,
+    payments,
+    post_story,
+    resilient_payment,
+    user,
+)
 from fastapi_cloudflow import attach_to_fastapi
 
 
 def create_app() -> FastAPI:
     app = FastAPI()
     # Import types used for stub endpoints
-    from flows.payments import PSPReq, PSPRes
-    from flows.user import IdentityReq, IdentityRes
+    from examples.app.flows.payments import PSPReq, PSPRes
+    from examples.app.flows.user import IdentityReq, IdentityRes
 
     @app.get("/health")
     def health() -> dict[str, str]:  # noqa: D401