Add routers to our fastapi app for testing

Author: bitterpanda63

sample-apps/fastapi-postgres-uvicorn/app.py

@@ -8,6 +8,7 @@
 from fastapi.templating import Jinja2Templates
 from fastapi.middleware.cors import CORSMiddleware
 from aikido_zen.middleware import AikidoFastAPIMiddleware
+from routers import dogs, admin, health
 
 templates = Jinja2Templates(directory="templates")
 
@@ -23,6 +24,10 @@
 )
 app.add_middleware(AikidoFastAPIMiddleware)
 
+app.include_router(dogs.router)
+app.include_router(admin.router)
+app.include_router(health.router)
+
 async def get_db_connection():
     return await asyncpg.connect(
         host="localhost",

sample-apps/fastapi-postgres-uvicorn/routers/__init__.py

@@ -0,0 +1,33 @@
+import asyncpg
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+router = APIRouter(prefix="/admin", tags=["admin"])
+
+
+async def get_db_connection():
+    return await asyncpg.connect(
+        host="localhost",
+        database="db",
+        user="user",
+        password="password",
+    )
+
+
+@router.get("/dogs")
+async def list_admin_dogs():
+    conn = await get_db_connection()
+    dogs = await conn.fetch("SELECT * FROM dogs WHERE isAdmin = TRUE")
+    await conn.close()
+    return JSONResponse({"admin_dogs": [dict(d) for d in dogs]})
+
+
+@router.delete("/dogs/{dog_id}")
+async def delete_dog(dog_id: int):
+    conn = await get_db_connection()
+    result = await conn.execute("DELETE FROM dogs WHERE id = $1", dog_id)
+    await conn.close()
+    deleted = int(result.split()[-1])
+    if deleted == 0:
+        return JSONResponse({"error": "Dog not found"}, status_code=404)
+    return JSONResponse({"message": f"Dog {dog_id} deleted"})

sample-apps/fastapi-postgres-uvicorn/routers/admin.py

@@ -0,0 +1,59 @@
+import asyncpg
+from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.templating import Jinja2Templates
+
+router = APIRouter(prefix="/dogs", tags=["dogs"])
+
+templates = Jinja2Templates(directory="templates")
+
+
+async def get_db_connection():
+    return await asyncpg.connect(
+        host="localhost",
+        database="db",
+        user="user",
+        password="password",
+    )
+
+
+@router.get("/", response_class=HTMLResponse)
+async def list_dogs(request: Request):
+    conn = await get_db_connection()
+    dogs = await conn.fetch("SELECT * FROM dogs")
+    await conn.close()
+    return templates.TemplateResponse(
+        "index.html", {"request": request, "title": "Dogs", "dogs": dogs}
+    )
+
+
+@router.get("/{dog_id}", response_class=HTMLResponse)
+async def get_dog(request: Request, dog_id: int):
+    conn = await get_db_connection()
+    dog = await conn.fetchrow("SELECT * FROM dogs WHERE id = $1", dog_id)
+    await conn.close()
+    if dog is None:
+        raise HTTPException(status_code=404, detail="Dog not found")
+    return templates.TemplateResponse(
+        "dogpage.html",
+        {"request": request, "title": "Dog", "dog": dog, "isAdmin": "Yes" if dog[2] else "No"},
+    )
+
+
+@router.post("/")
+async def create_dog(request: Request):
+    data = await request.form()
+    dog_name = data.get("dog_name")
+
+    if not dog_name:
+        return JSONResponse({"error": "dog_name is required"}, status_code=400)
+
+    conn = await get_db_connection()
+    try:
+        await conn.execute(
+            "INSERT INTO dogs (dog_name, isAdmin) VALUES ($1, FALSE)", dog_name
+        )
+    finally:
+        await conn.close()
+
+    return JSONResponse({"message": f"Dog {dog_name} created successfully"}, status_code=201)

sample-apps/fastapi-postgres-uvicorn/routers/dogs.py

@@ -0,0 +1,14 @@
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+router = APIRouter(prefix="/health", tags=["health"])
+
+
+@router.get("/")
+async def health():
+    return JSONResponse({"status": "ok"})
+
+
+@router.get("/ping")
+async def ping():
+    return JSONResponse({"ping": "pong"})