Complete FastAPI refactoring - organize code into modular structure

Co-authored-by: ks6088ts <[email protected]>

Author: Copilot

template_fastapi/app.py

@@ -3,20 +3,16 @@
 ref. https://github.com/tadata-org/fastapi_mcp/blob/v0.3.4/examples/shared/apps/items.py
 """
 
-import random
 import uuid
 from os import getenv
 
 from azure.monitor.opentelemetry import configure_azure_monitor
-from fastapi import FastAPI, HTTPException, Query
+from fastapi import FastAPI
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 from opentelemetry.trace import Span
-from pydantic import BaseModel
 
-from template_fastapi.opentelemetry import get_meter, get_tracer
+from template_fastapi.routers import demo, games, items
 
-tracer = get_tracer(__name__)
-meter = get_meter(__name__)
 app = FastAPI()
 
 # If APPLICATIONINSIGHTS_CONNECTION_STRING exists, configure Azure Monitor
@@ -39,201 +35,7 @@ def server_request_hook(span: Span, scope: dict):
     )
     FastAPIInstrumentor.instrument_app(app)
 
-
-class Item(BaseModel):
-    id: int
-    name: str
-    description: str | None = None
-    price: float
-    tags: list[str] = []
-
-
-items_db: dict[int, Item] = {}
-
-
-@app.get("/items/", response_model=list[Item], tags=["items"], operation_id="list_items")
-async def list_items(skip: int = 0, limit: int = 10):
-    """
-    List all items in the database.
-
-    Returns a list of items, with pagination support.
-    """
-    return list(items_db.values())[skip : skip + limit]
-
-
-@app.get("/items/{item_id}", response_model=Item, tags=["items"], operation_id="get_item")
-async def read_item(item_id: int):
-    """
-    Get a specific item by its ID.
-
-    Raises a 404 error if the item does not exist.
-    """
-    if item_id not in items_db:
-        raise HTTPException(status_code=404, detail="Item not found")
-    return items_db[item_id]
-
-
-@app.post("/items/", response_model=Item, tags=["items"], operation_id="create_item")
-async def create_item(item: Item):
-    """
-    Create a new item in the database.
-
-    Returns the created item with its assigned ID.
-    """
-    items_db[item.id] = item
-    return item
-
-
-@app.put("/items/{item_id}", response_model=Item, tags=["items"], operation_id="update_item")
-async def update_item(item_id: int, item: Item):
-    """
-    Update an existing item.
-
-    Raises a 404 error if the item does not exist.
-    """
-    if item_id not in items_db:
-        raise HTTPException(status_code=404, detail="Item not found")
-
-    item.id = item_id
-    items_db[item_id] = item
-    return item
-
-
-@app.delete("/items/{item_id}", tags=["items"], operation_id="delete_item")
-async def delete_item(item_id: int):
-    """
-    Delete an item from the database.
-
-    Raises a 404 error if the item does not exist.
-    """
-    if item_id not in items_db:
-        raise HTTPException(status_code=404, detail="Item not found")
-
-    del items_db[item_id]
-    return {"message": "Item deleted successfully"}
-
-
-@app.get("/items/search/", response_model=list[Item], tags=["search"], operation_id="search_items")
-async def search_items(
-    q: str | None = Query(None, description="Search query string"),
-    min_price: float | None = Query(None, description="Minimum price"),
-    max_price: float | None = Query(None, description="Maximum price"),
-    tags: list[str] = Query([], description="Filter by tags"),
-):
-    """
-    Search for items with various filters.
-
-    Returns a list of items that match the search criteria.
-    """
-    results = list(items_db.values())
-
-    if q:
-        q = q.lower()
-        results = [
-            item
-            for item in results
-            if q in item.name.lower() or (item.description is not None and q in item.description.lower())
-        ]
-
-    if min_price is not None:
-        results = [item for item in results if item.price >= min_price]
-    if max_price is not None:
-        results = [item for item in results if item.price <= max_price]
-
-    if tags:
-        results = [item for item in results if all(tag in item.tags for tag in tags)]
-
-    return results
-
-
-sample_items = [
-    Item(id=1, name="Hammer", description="A tool for hammering nails", price=9.99, tags=["tool", "hardware"]),
-    Item(id=2, name="Screwdriver", description="A tool for driving screws", price=7.99, tags=["tool", "hardware"]),
-    Item(id=3, name="Wrench", description="A tool for tightening bolts", price=12.99, tags=["tool", "hardware"]),
-    Item(id=4, name="Saw", description="A tool for cutting wood", price=19.99, tags=["tool", "hardware", "cutting"]),
-    Item(id=5, name="Drill", description="A tool for drilling holes", price=49.99, tags=["tool", "hardware", "power"]),
-]
-for item in sample_items:
-    items_db[item.id] = item
-
-
-# Add flaky API which receives percentage of failure
-@app.get("/flaky/{failure_rate}", tags=["flaky"], operation_id="flaky")
-async def flaky(failure_rate: int):
-    """
-    A flaky endpoint that simulates a failure based on the provided failure rate.
-
-    The failure rate is a percentage (0-100) that determines the likelihood of failure.
-    """
-    if not (0 <= failure_rate <= 100):
-        raise HTTPException(
-            status_code=400,
-            detail="Failure rate must be between 0 and 100",
-        )
-
-    if random.randint(0, 100) < failure_rate:
-        raise HTTPException(
-            status_code=500,
-            detail="Simulated failure",
-        )
-
-    return {
-        "message": "Request succeeded",
-    }
-
-
-# Add flaky API which raises an exception
-@tracer.start_as_current_span("flaky_exception")
-@app.get("/flaky/exception", tags=["flaky"], operation_id="flaky_exception")
-async def flaky_exception():
-    """
-    A flaky endpoint that always raises an exception.
-    """
-    raise HTTPException(
-        status_code=500,
-        detail="Simulated exception",
-    )
-
-
-# Add a heavy synchronous endpoint which receives milliseconds to sleep
-@app.get("/heavy_sync/{sleep_ms}", tags=["heavy"], operation_id="heavy_sync_with_sleep")
-async def heavy_sync_with_sleep(sleep_ms: int):
-    """
-    A heavy synchronous endpoint that sleeps for the specified number of milliseconds.
-
-    This simulates a long-running synchronous operation.
-    """
-    if sleep_ms < 0:
-        raise HTTPException(
-            status_code=400,
-            detail="Sleep time must be a non-negative integer",
-        )
-
-    import time
-
-    with tracer.start_as_current_span("parent"):
-        print(f"Sleeping for {sleep_ms} milliseconds")
-        time.sleep(sleep_ms / 1000.0)
-        with tracer.start_as_current_span("child"):
-            print("Child span")
-    return {
-        "message": f"Slept for {sleep_ms} milliseconds",
-    }
-
-
-# Add counter for dice rolls
-roll_counter = meter.create_counter(
-    "dice.rolls",
-    description="The number of rolls by roll value",
-)
-
-
-@app.get("/roll_dice", operation_id="roll_dice")
-async def roll_dice():
-    """
-    Simulate rolling a dice and record the roll in the meter.
-    """
-    with tracer.start_as_current_span("roll_dice"):
-        roll = random.randint(1, 6)
-        roll_counter.add(1, {"roll.value": str(roll)})
-    return roll
+# Include routers
+app.include_router(items.router)
+app.include_router(demo.router)
+app.include_router(games.router)

template_fastapi/database/__init__.py

@@ -0,0 +1 @@
+# Database module for FastAPI application

template_fastapi/database/items.py

@@ -0,0 +1,17 @@
+from template_fastapi.models.item import Item
+
+# In-memory database for items
+items_db: dict[int, Item] = {}
+
+# Sample data initialization
+sample_items = [
+    Item(id=1, name="Hammer", description="A tool for hammering nails", price=9.99, tags=["tool", "hardware"]),
+    Item(id=2, name="Screwdriver", description="A tool for driving screws", price=7.99, tags=["tool", "hardware"]),
+    Item(id=3, name="Wrench", description="A tool for tightening bolts", price=12.99, tags=["tool", "hardware"]),
+    Item(id=4, name="Saw", description="A tool for cutting wood", price=19.99, tags=["tool", "hardware", "cutting"]),
+    Item(id=5, name="Drill", description="A tool for drilling holes", price=49.99, tags=["tool", "hardware", "power"]),
+]
+
+# Initialize database with sample data
+for item in sample_items:
+    items_db[item.id] = item

template_fastapi/models/__init__.py

@@ -0,0 +1 @@
+# Models module for FastAPI application

template_fastapi/models/item.py

@@ -0,0 +1,9 @@
+from pydantic import BaseModel
+
+
+class Item(BaseModel):
+    id: int
+    name: str
+    description: str | None = None
+    price: float
+    tags: list[str] = []

template_fastapi/routers/__init__.py

@@ -0,0 +1 @@
+# Routers module for FastAPI application

template_fastapi/routers/demo.py

@@ -0,0 +1,68 @@
+import random
+import time
+
+from fastapi import APIRouter, HTTPException
+
+from template_fastapi.opentelemetry import get_tracer
+
+tracer = get_tracer(__name__)
+router = APIRouter()
+
+
+@tracer.start_as_current_span("flaky_exception")
+@router.get("/flaky/exception", tags=["flaky"], operation_id="flaky_exception")
+async def flaky_exception():
+    """
+    A flaky endpoint that always raises an exception.
+    """
+    raise HTTPException(
+        status_code=500,
+        detail="Simulated exception",
+    )
+
+
+@router.get("/flaky/{failure_rate}", tags=["flaky"], operation_id="flaky")
+async def flaky(failure_rate: int):
+    """
+    A flaky endpoint that simulates a failure based on the provided failure rate.
+
+    The failure rate is a percentage (0-100) that determines the likelihood of failure.
+    """
+    if not (0 <= failure_rate <= 100):
+        raise HTTPException(
+            status_code=400,
+            detail="Failure rate must be between 0 and 100",
+        )
+
+    if random.randint(0, 100) < failure_rate:
+        raise HTTPException(
+            status_code=500,
+            detail="Simulated failure",
+        )
+
+    return {
+        "message": "Request succeeded",
+    }
+
+
+@router.get("/heavy_sync/{sleep_ms}", tags=["heavy"], operation_id="heavy_sync_with_sleep")
+async def heavy_sync_with_sleep(sleep_ms: int):
+    """
+    A heavy synchronous endpoint that sleeps for the specified number of milliseconds.
+
+    This simulates a long-running synchronous operation.
+    """
+    if sleep_ms < 0:
+        raise HTTPException(
+            status_code=400,
+            detail="Sleep time must be a non-negative integer",
+        )
+
+    with tracer.start_as_current_span("parent"):
+        print(f"Sleeping for {sleep_ms} milliseconds")
+        time.sleep(sleep_ms / 1000.0)
+        with tracer.start_as_current_span("child"):
+            print("Child span")
+    return {
+        "message": f"Slept for {sleep_ms} milliseconds",
+    }

template_fastapi/routers/games.py

@@ -0,0 +1,27 @@
+import random
+
+from fastapi import APIRouter
+
+from template_fastapi.opentelemetry import get_meter, get_tracer
+
+tracer = get_tracer(__name__)
+meter = get_meter(__name__)
+
+# Add counter for dice rolls
+roll_counter = meter.create_counter(
+    "dice.rolls",
+    description="The number of rolls by roll value",
+)
+
+router = APIRouter()
+
+
+@router.get("/roll_dice", operation_id="roll_dice")
+async def roll_dice():
+    """
+    Simulate rolling a dice and record the roll in the meter.
+    """
+    with tracer.start_as_current_span("roll_dice"):
+        roll = random.randint(1, 6)
+        roll_counter.add(1, {"roll.value": str(roll)})
+    return roll