Add example for medium/large projects in FastAPI (#595)

examples/fastapi/api/__init__.py

@@ -0,0 +1,7 @@
+from fastapi import APIRouter
+
+from api.my_api import router as my_api_router
+
+router = APIRouter()
+
+router.include_router(my_api_router)

examples/fastapi/api/my_api.py

@@ -0,0 +1,16 @@
+from typing import Union
+from fastapi import APIRouter
+
+from my_faust.table.my_table import greetings_table
+
+router = APIRouter()
+
+
+@router.get("/items/{item_id}")
+def read_item(item_id: int, q: Union[str, None] = None):
+    return {"item_id": item_id, "q": q}
+
+
+@router.get("/table")
+def read_table():
+    return [{k: v} for k, v in greetings_table.items()]

examples/fastapi/main.py

@@ -0,0 +1,42 @@
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from api import router as api_router
+
+from my_faust.timer import router as timer_router
+from my_faust.app import faust_app
+
+
+# This is just hello_world.py integrated with a FastAPI application
+
+
+def fake_answer_to_everything_ml_model(x: float):
+    return x * 42
+
+
+ml_models = {}
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    faust_app.discover()
+    await faust_app.start()
+    yield
+    await faust_app.stop()
+
+
+# You MUST have "app" defined in order for Faust to discover the app
+# if you're using "faust" on CLI, but this doesn't work yet
+app = fastapi_app = FastAPI(
+    lifespan=lifespan,
+)
+
+# For now, run via "uvicorn fastapi_example:app"
+# then visit http://127.0.0.1:8000/docs
+
+app.include_router(router=api_router)
+app.include_router(router=timer_router)
+
+
+@app.get("/")
+def read_root():
+    return {"Hello": "World"}

examples/fastapi/my_faust/agent/my_agent.py

@@ -0,0 +1,10 @@
+from my_faust.app import faust_app
+from my_faust.table.my_table import greetings_table
+from my_faust.topic.my_topic import greetings_topic
+
+
+@faust_app.agent(greetings_topic)
+async def print_greetings(greetings):
+    async for greeting in greetings:
+        print(f"greeting: {greeting}")
+        greetings_table[greeting] = {"hello world"}

examples/fastapi/my_faust/app.py

@@ -0,0 +1,18 @@
+import faust
+
+
+def get_all_packages_to_scan():
+    return ["my_faust"]
+
+
+# You MUST have "app" defined in order for Faust to discover the app
+# if you're using "faust" on CLI, but this doesn't work yet
+# autodiscover https://faust-streaming.github.io/faust/userguide/settings.html#autodiscover
+app = faust_app = faust.App(
+    'hello-world-fastapi',
+    broker='kafka://localhost:9092',
+    web_enabled=False,
+    autodiscover=get_all_packages_to_scan,
+)
+
+# For now, run via "faust -A my_faust.app worker -l info"

examples/fastapi/my_faust/table/my_table.py

@@ -0,0 +1,8 @@
+from my_faust.app import faust_app
+
+greetings_table = faust_app.GlobalTable(
+    name="greetings_table",
+    default=dict,
+    partitions=1,
+    recovery_buffer_size=1,
+)

examples/fastapi/my_faust/timer/__init__.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+from my_faust.timer.my_timer import router as my_timer_router
+
+router = APIRouter()
+
+router.include_router(my_timer_router)

examples/fastapi/my_faust/timer/my_timer.py

@@ -0,0 +1,14 @@
+from uuid import uuid4
+from fastapi import APIRouter
+
+from my_faust.app import faust_app
+from my_faust.topic.my_topic import greetings_topic
+
+router = APIRouter()
+
+
+@faust_app.timer(5)  # make sure you *always* add the timer above if you're using one
+@router.get("/produce")
+async def produce():
+    await greetings_topic.send(value=uuid4().hex)
+    return {"success": True}

examples/fastapi/my_faust/topic/my_topic.py

@@ -0,0 +1,3 @@
+from my_faust.app import faust_app
+
+greetings_topic = faust_app.topic("greetings", value_type=str)