better examples

Co-authored-by: Copilot <copilot@github.com>

Author: codeSamuraii

examples/cancellation.py

@@ -21,16 +21,15 @@
 @trace
 async def slow_computation(n: int) -> int:
     """A deliberately slow function."""
+    total = 0
     try:
-        total = 0
         for _ in range(n):
             await asyncio.sleep(1.0)
             total += 1
         return total
     finally:
         if total < n:
-            print("Task cancelled")
-
+            print(f"Task interrupted after {total}/{n} iterations")
 
 
 async def main() -> None:
@@ -46,6 +45,7 @@ async def main() -> None:
     # Awaiting a cancelled task raises TaskCancelled
     try:
         result = await future
+        print(f"Unexpected result: {result}")
     except TaskCancelled:
         print("Task was cancelled successfully!")
 
@@ -54,4 +54,5 @@ async def main() -> None:
     print(f"Task status: {status}")  # "cancelled"
 
 
-asyncio.run(main())
+if __name__ == "__main__":
+    asyncio.run(main())

examples/email_attachments.py

@@ -1,4 +1,4 @@
-"""Poll an IMAP mailbox locally, fan attachment work out to workers.
+"""Inspect email attachments on workers, fanned out from a local poller.
 
 Two concerns kept apart:
 
@@ -14,15 +14,18 @@
 
 Usage:
     pyfuse worker --backend redis://localhost:6379 --tmp
-    python -m pyfuse run --tmp examples/email_attachments.py
+    python examples/email_attachments.py
+
+The script synthesizes a handful of emails with attachments in-memory
+(no IMAP server needed), dispatches each attachment to the worker, and
+prints the verdict.
 """
 
 import asyncio
 import hashlib
-import imaplib
 import re
 from email import message_from_bytes
-from email.message import Message
+from email.message import EmailMessage, Message
 from typing import Any
 
 import pyfuse
@@ -84,7 +87,7 @@ def process_attachment(filename: str, payload: bytes) -> dict[str, Any]:
     }
 
 
-# --- local-only IMAP plumbing --------------------------------------------
+# --- local-only mailbox plumbing -----------------------------------------
 
 def iter_attachments(msg: Message) -> list[tuple[str, bytes]]:
     out: list[tuple[str, bytes]] = []
@@ -100,45 +103,59 @@ def iter_attachments(msg: Message) -> list[tuple[str, bytes]]:
     return out
 
 
-def fetch_unseen(host: str, user: str, password: str) -> list[Message]:
-    cli = imaplib.IMAP4_SSL(host)
-    try:
-        cli.login(user, password)
-        cli.select("INBOX")
-        _, data = cli.search(None, "UNSEEN")
-        msgs: list[Message] = []
-        for uid in data[0].split():
-            _, parts = cli.fetch(uid, "(RFC822)")
-            raw = parts[0][1] if parts and parts[0] else None
-            if isinstance(raw, bytes):
-                msgs.append(message_from_bytes(raw))
-        return msgs
-    finally:
-        cli.logout()
+def synthesize_inbox() -> list[Message]:
+    """Build a small in-memory inbox with attachments for the demo."""
+    samples: list[tuple[str, str, list[tuple[str, bytes]]]] = [
+        (
+            "alice@example.com",
+            "Lunch tomorrow?",
+            [("photo.jpg", b"\xff\xd8\xff" + b"\x00" * 4096)],
+        ),
+        (
+            "billing@vendor.com",
+            "Invoice INV-9921",
+            [("invoice.pdf", b"%PDF-1.4 invoice total: $1,250.00 due upon receipt")],
+        ),
+        (
+            "noreply@scary.example",
+            "URGENT: please review attached",
+            [("urgent_invoice.scr", b"MZ" + b"\x00" * 2048)],
+        ),
+        (
+            "ops@partner.com",
+            "Q1 deck and notes",
+            [
+                ("deck.pdf", b"%PDF-1.4 strategy review Q1"),
+                ("notes.txt", b"meeting notes go here"),
+            ],
+        ),
+    ]
+
+    inbox: list[Message] = []
+    for sender, subject, atts in samples:
+        msg = EmailMessage()
+        msg["From"] = sender
+        msg["To"] = "you@example.com"
+        msg["Subject"] = subject
+        msg.set_content("See attached.")
+        for name, payload in atts:
+            msg.add_attachment(
+                payload, maintype="application", subtype="octet-stream", filename=name,
+            )
+        inbox.append(message_from_bytes(bytes(msg)))
+    return inbox
 
 
 async def main() -> None:
-    host, user, password = "imap.example.com", "you@example.com", "secret"
-
-    while True:
-        try:
-            messages = await asyncio.to_thread(fetch_unseen, host, user, password)
-        except Exception as exc:
-            print(f"poll failed: {exc}")
-            await asyncio.sleep(30)
-            continue
-
-        attachments = [a for m in messages for a in iter_attachments(m)]
-        if not attachments:
-            await asyncio.sleep(30)
-            continue
-
-        results = await process_attachment.map(attachments)
-        for r in results:
-            flag = "!!" if r["risk"] >= 50 else "ok"
-            print(f"  [{flag}] {r['filename']:<40} {r['kind']:<10} risk={r['risk']}")
-
-        await asyncio.sleep(30)
+    messages = synthesize_inbox()
+    attachments = [a for m in messages for a in iter_attachments(m)]
+    print(f"Inbox: {len(messages)} messages, {len(attachments)} attachments")
+
+    results = await process_attachment.map(attachments)
+    for r in results:
+        flag = "!!" if r["risk"] >= 50 else "ok"
+        print(f"  [{flag}] {r['filename']:<24} {r['kind']:<10} "
+              f"size={r['size']:<6} risk={r['risk']}")
 
 
 if __name__ == "__main__":

examples/image_thumbnails.py

@@ -14,6 +14,14 @@
     python examples/large_module.py
 """
 import asyncio
+import sys
+from pathlib import Path
+
+# Make ``tests.fixtures.stress_test_module`` importable when running this
+# script directly from a checkout, without needing PYTHONPATH gymnastics.
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
 
 import pyfuse
 from pyfuse import trace
@@ -47,18 +55,13 @@ def full_sensor_report(sensor_count: int, readings_per_sensor: int, seed: int =
         stats, anomalies, normalized,
     )
     report["delta_count"] = len(deltas)
-    r = format_text_report(report)
-    print(r)
-    return r
+    return format_text_report(report)
+
 
 async def main() -> None:
-    # Connect to the worker
     pyfuse.connect("redis://localhost:6379")
-
-    # Remote call (same function, same args, on a worker)
-    remote_result = await full_sensor_report.run(3, 50, seed=42)
-
-    print(f"Success! Local and remote results match:\n  {repr(remote_result)[:80] + '...'}")
+    report = await full_sensor_report.run(3, 50, seed=42)
+    print(report)
 
 
 if __name__ == "__main__":

examples/large_module.py

@@ -15,15 +15,20 @@
 
 Usage:
     pyfuse worker --backend redis://localhost:6379 --tmp
-    uvicorn examples.pdf_report:app --reload
+    python examples/pdf_report.py
 
-    curl -X POST localhost:8000/reports -o report.pdf \\
-         -H 'content-type: application/json' \\
-         -d '{"title":"Q1","rows":[["Revenue",120000],["Costs",80000]]}'
+The script starts the FastAPI app in-process, posts a sample report,
+prints the resulting PDF size (and writes it to ``/tmp/pyfuse_report.pdf``),
+and exits.
 """
 
+import asyncio
+from contextlib import asynccontextmanager
 from io import BytesIO
+from pathlib import Path
+from typing import AsyncIterator
 
+import httpx
 import uvicorn
 from fastapi import FastAPI
 from fastapi.responses import Response
@@ -68,32 +73,61 @@ def render_report(title: str, rows: list[list[str | float]]) -> bytes:
 # --- FastAPI app ----------------------------------------------------------
 
 class ReportRequest(BaseModel):
-    title: str = "Example Report"
+    title: str = "Quarterly Report"
     rows: list[list[str | float]] = [
         ["Revenue", 120000],
         ["Costs", 80000],
+        ["Profit", 40000],
     ]
 
 
-app = FastAPI(title="pyfuse PDF service")
-
-
-@app.on_event("startup")
-async def _startup() -> None:
+@asynccontextmanager
+async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     pyfuse.connect("redis://localhost:6379")
+    try:
+        yield
+    finally:
+        await pyfuse.disconnect()
 
 
-@app.on_event("shutdown")
-async def _shutdown() -> None:
-    await pyfuse.disconnect()
+app = FastAPI(title="pyfuse PDF service", lifespan=lifespan)
 
 
-@app.get("/reports")
-async def make_report() -> Response:
-    req = ReportRequest()
+@app.post("/reports")
+async def make_report(req: ReportRequest) -> Response:
     pdf = await render_report.run(req.title, req.rows)
     return Response(content=pdf, media_type="application/pdf")
 
 
+# --- self-driving demo ----------------------------------------------------
+
+async def _demo() -> None:
+    config = uvicorn.Config(app, host="127.0.0.1", port=8080, log_level="warning")
+    server = uvicorn.Server(config)
+    task = asyncio.create_task(server.serve())
+    while not server.started:
+        await asyncio.sleep(0.5)
+
+    payload = {
+        "title": "Q1 2026 Report",
+        "rows": [
+            ["Revenue", 240_000],
+            ["Cost of goods", 90_000],
+            ["Operating expenses", 55_000],
+            ["Net profit", 95_000],
+        ],
+    }
+    try:
+        async with httpx.AsyncClient(base_url="http://127.0.0.1:8080") as client:
+            resp = await client.post("/reports", json=payload, timeout=60.0)
+            resp.raise_for_status()
+            out = Path("/tmp/pyfuse_report.pdf")
+            out.write_bytes(resp.content)
+            print(f"PDF: {len(resp.content)} bytes -> {out}")
+    finally:
+        server.should_exit = True
+        await task
+
+
 if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=8080)
+    asyncio.run(_demo())