feat(examples): add playful flows and FastAPI app wiring

Author: joaodaher

examples/__init__.py

@@ -0,0 +1,26 @@
+FROM python:3.13-slim
+
+# Install fastapi-cloudflow from local package sources (only what's needed)
+WORKDIR /opt/fastapi-cloudflow
+COPY pyproject.toml ./
+COPY README.md ./
+COPY src/ ./src/
+RUN pip install --no-cache-dir .
+
+# Set up the application directory with only the app code
+WORKDIR /app
+COPY examples/app/ .
+
+# Install application dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Ensure the app root is on Python import path for CLI/tools and make Python friendlier in containers
+ENV PYTHONPATH=/app \
+    PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+# Expose port (Cloud Run will override this with PORT env var)
+EXPOSE 8080
+
+# Run the application using PORT environment variable (Cloud Run standard)
+CMD uvicorn main:app --host 0.0.0.0 --port ${PORT:-8080}

examples/app/Dockerfile

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from fastapi_cloudflow import Arg, Context, HttpStep, step, workflow
+
+
+class EchoIn(BaseModel):
+    name: str
+
+
+class HttpbinRes(BaseModel):
+    # Avoid shadowing BaseModel methods by using an alias
+    model_config = {"populate_by_name": True}
+    json_: dict = Field(alias="json")
+
+
+class EchoOut(BaseModel):
+    echoed_name: str
+
+
+class NameShout(BaseModel):
+    name_upper: str
+    length: int
+
+
+httpbin_echo = HttpStep(
+    name="external-echo",
+    input_model=EchoIn,
+    output_model=HttpbinRes,
+    method="POST",
+    # Use a public echo endpoint (httpbin) via env, default set at deploy
+    url=Arg.env("ECHO_URL"),
+)
+
+
+@step(name="extract-name")
+async def adapt_echo(ctx: Context, data: HttpbinRes) -> EchoOut:
+    # httpbin returns payload under the "json" key
+    name = str(data.json_.get("name", ""))
+    return EchoOut(echoed_name=name)
+
+
+@step(name="name-shout")
+async def name_shout(ctx: Context, data: EchoOut) -> NameShout:
+    up = data.echoed_name.upper()
+    return NameShout(name_upper=up, length=len(up))
+
+
+ECHO_NAME_FLOW = (workflow("echo-name-flow") >> httpbin_echo >> adapt_echo >> name_shout).build()

examples/app/__init__.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from fastapi_cloudflow import Context, HttpStep, step, workflow
+
+
+class EmptyIn(BaseModel):
+    pass
+
+
+class JokeAPIRes(BaseModel):
+    id: str
+    joke: str
+    status: int
+
+
+class JokeBits(BaseModel):
+    setup: str
+    punch: str
+
+
+joke_fetch = HttpStep(
+    name="joke-fetch",
+    input_model=EmptyIn,
+    output_model=JokeAPIRes,
+    method="GET",
+    url="https://icanhazdadjoke.com/",
+    headers={"Accept": "application/json"},
+)
+
+
+@step(name="joke-split")
+async def joke_split(ctx: Context, data: JokeAPIRes) -> JokeBits:
+    j = data.joke or ""
+    mid = len(j) // 2
+    return JokeBits(setup=j[:mid], punch=j[mid:])
+
+
+class JokeRated(BaseModel):
+    setup: str
+    punch: str
+    rating: int = Field(ge=0, le=10)
+
+
+@step(name="joke-rate")
+async def joke_rate(ctx: Context, data: JokeBits) -> JokeRated:
+    rating = (len(data.setup) + len(data.punch)) % 11
+    return JokeRated(setup=data.setup, punch=data.punch, rating=rating)
+
+
+JOKE_FLOW = (workflow("joke-flow") >> joke_fetch >> joke_split >> joke_rate).build()

examples/app/flows/echo_name.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from decimal import Decimal
+from uuid import uuid4
+
+from pydantic import BaseModel
+
+from fastapi_cloudflow import Context, step, workflow
+
+
+class CreateOrder(BaseModel):
+    account_id: int
+    sku: str
+    qty: int
+
+
+class OrderDraft(BaseModel):
+    order_id: str
+    price: Decimal
+
+
+class PaymentAuth(BaseModel):
+    order_id: str
+    status: str
+
+
+@step(name="price-order")
+async def price_order(ctx: Context, data: CreateOrder) -> OrderDraft:
+    return OrderDraft(order_id=f"o-{uuid4().hex}", price=Decimal("12.34"))
+
+
+@step(name="auth-payment")
+async def auth_payment(ctx: Context, data: OrderDraft) -> PaymentAuth:
+    return PaymentAuth(order_id=data.order_id, status="approved")
+
+
+ORDER_FLOW = (workflow("order-flow") >> price_order >> auth_payment).build()

examples/app/flows/jokes.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from fastapi_cloudflow import Arg, AssignStep, Context, HttpStep, step, workflow
+
+
+class Cart(BaseModel):
+    total: float
+    currency: str
+
+
+class PSPReq(BaseModel):
+    amount: float
+    currency: str
+
+
+class PSPRes(BaseModel):
+    status: str
+    psp_id: str
+
+
+@step(name="validate-cart")
+async def validate_cart(ctx: Context, data: Cart) -> Cart:
+    assert data.total >= 0
+    return data
+
+
+to_psp = AssignStep("cart->psp", Cart, PSPReq, expr={"amount": "${payload.total}", "currency": "${payload.currency}"})
+
+psp = HttpStep(
+    "psp-charge",
+    PSPReq,
+    PSPRes,
+    method="POST",
+    url=Arg.env("PSP_URL") / "charge",
+    auth={"type": "OIDC", "audience": Arg.env("PSP_URL")},
+)
+
+
+class ChargeResult(BaseModel):
+    ok: bool
+    txn_id: str | None
+
+
+@step(name="summarize-charge")
+async def summarize_charge(ctx: Context, data: PSPRes) -> ChargeResult:
+    return ChargeResult(ok=(data.status == "approved"), txn_id=data.psp_id)
+
+
+PAYMENT_FLOW = (workflow("payment-flow") >> validate_cart >> to_psp >> psp >> summarize_charge).build()

examples/app/flows/order.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from fastapi_cloudflow import Context, HttpStep, step, workflow
+
+
+class StoryIn(BaseModel):
+    topic: str
+    mood: str
+    author_id: int
+
+
+class PostOut(BaseModel):
+    id: int
+    title: str
+    body: str
+    userId: int
+
+
+class PostSummary(BaseModel):
+    id: int
+    slug: str
+    short_title: str
+
+
+@step(name="build-story")
+async def build_story(ctx: Context, data: StoryIn) -> PostOut:
+    title = f"{data.topic} in a {data.mood} mood"
+    body = f"Once upon a time about {data.topic}"
+    return PostOut(id=0, title=title, body=body, userId=data.author_id)
+
+
+jp_create_post = HttpStep(
+    name="create-post",
+    input_model=PostOut,
+    output_model=PostOut,
+    method="POST",
+    url="https://jsonplaceholder.typicode.com/posts",
+)
+
+
+@step(name="summarize-post")
+async def summarize_post(ctx: Context, data: PostOut) -> PostSummary:
+    slug = f"{data.userId}-{(data.title or '').lower().replace(' ', '-')[:24]}"
+    short = (data.title or "")[:16]
+    return PostSummary(id=data.id, slug=slug, short_title=short)
+
+
+POST_STORY_FLOW = (workflow("post-story-flow") >> build_story >> jp_create_post >> summarize_post).build()

examples/app/flows/payments.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, EmailStr, Field
+
+from fastapi_cloudflow import Arg, AssignStep, Context, HttpStep, step, workflow
+
+
+class SignupRequest(BaseModel):
+    email: EmailStr
+    password: str = Field(min_length=8)
+
+
+class UserDraft(BaseModel):
+    email: EmailStr
+    hashed_password: str
+
+
+class IdentityReq(BaseModel):
+    email: EmailStr
+    secret: str
+
+
+class IdentityRes(BaseModel):
+    external_id: str
+    ok: bool
+
+
+class UserCreated(BaseModel):
+    user_id: str
+    email: EmailStr
+
+
+@step(name="hash-password")
+async def hash_password(ctx: Context, data: SignupRequest) -> UserDraft:
+    return UserDraft(email=data.email, hashed_password=f"hashed:{data.password}")
+
+
+adapt_to_idp = AssignStep(
+    "draft->idp",
+    UserDraft,
+    IdentityReq,
+    expr={
+        "email": "${payload.email}",
+        "secret": "${payload.hashed_password}",
+    },
+)
+
+
+idp_call = HttpStep(
+    name="idp-call",
+    input_model=IdentityReq,
+    output_model=IdentityRes,
+    method="POST",
+    url=Arg.env("IDP_URL") / "signup",
+    auth={"type": "OIDC", "audience": Arg.env("IDP_URL")},
+)
+
+
+@step(name="persist-user")
+async def persist_user(ctx: Context, data: IdentityRes) -> UserCreated:
+    return UserCreated(user_id=data.external_id, email="user@example.com")
+
+
+USER_SIGNUP = (workflow("user-signup") >> hash_password >> adapt_to_idp >> idp_call >> persist_user).build()

examples/app/flows/post_story.py

@@ -0,0 +1,30 @@
+from fastapi import FastAPI
+from flows import echo_name, jokes, order, payments, post_story, user  # noqa: F401
+
+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
+
+    @app.get("/health")
+    def health() -> dict[str, str]:  # noqa: D401
+        return {"status": "ok"}
+
+    # Simple stubs so example HttpSteps resolve successfully during smoke tests
+    @app.post("/signup", response_model=IdentityRes)
+    def signup(req: IdentityReq) -> IdentityRes:
+        return IdentityRes(external_id="u-0001", ok=True)
+
+    @app.post("/charge", response_model=PSPRes)
+    def charge(req: PSPReq) -> PSPRes:
+        return PSPRes(status="approved", psp_id="p-0001")
+
+    attach_to_fastapi(app)
+    return app
+
+
+app = create_app()