asgi/wsgi mounting implemented

examples/apps/starlette_app.py

@@ -0,0 +1,112 @@
+"""
+Example Starlette application for PyServe ASGI mounting.
+
+This demonstrates how to create a Starlette application that can be
+mounted at a specific path in PyServe.
+"""
+
+try:
+    from starlette.applications import Starlette
+    from starlette.responses import JSONResponse
+    from starlette.routing import Route
+    from starlette.requests import Request
+except ImportError:
+    raise ImportError(
+        "Starlette is not installed. Install with: pip install starlette"
+    )
+
+
+tasks_db = {
+    1: {"id": 1, "title": "Task 1", "completed": False},
+    2: {"id": 2, "title": "Task 2", "completed": True},
+}
+
+
+async def homepage(request: Request) -> JSONResponse:
+    return JSONResponse({
+        "message": "Welcome to Starlette mounted in PyServe!"
+    })
+
+
+async def health_check(request: Request) -> JSONResponse:
+    return JSONResponse({"status": "healthy", "app": "starlette"})
+
+
+async def list_tasks(request: Request) -> JSONResponse:
+    return JSONResponse({
+        "tasks": list(tasks_db.values()),
+        "count": len(tasks_db)
+    })
+
+
+async def get_task(request: Request) -> JSONResponse:
+    task_id = int(request.path_params["task_id"])
+    if task_id not in tasks_db:
+        return JSONResponse({"error": "Task not found"}, status_code=404)
+    return JSONResponse(tasks_db[task_id])
+
+
+async def create_task(request: Request) -> JSONResponse:
+    data = await request.json()
+    if not data or "title" not in data:
+        return JSONResponse({"error": "Title is required"}, status_code=400)
+
+    new_id = max(tasks_db.keys()) + 1 if tasks_db else 1
+    tasks_db[new_id] = {
+        "id": new_id,
+        "title": data["title"],
+        "completed": data.get("completed", False),
+    }
+    return JSONResponse(
+        {"message": f"Task created with ID {new_id}", "task": tasks_db[new_id]},
+        status_code=201
+    )
+
+
+async def update_task(request: Request) -> JSONResponse:
+    task_id = int(request.path_params["task_id"])
+    if task_id not in tasks_db:
+        return JSONResponse({"error": "Task not found"}, status_code=404)
+
+    data = await request.json()
+    if data:
+        if "title" in data:
+            tasks_db[task_id]["title"] = data["title"]
+        if "completed" in data:
+            tasks_db[task_id]["completed"] = data["completed"]
+
+    return JSONResponse({
+        "message": f"Task {task_id} updated",
+        "task": tasks_db[task_id]
+    })
+
+
+async def delete_task(request: Request) -> JSONResponse:
+    task_id = int(request.path_params["task_id"])
+    if task_id not in tasks_db:
+        return JSONResponse({"error": "Task not found"}, status_code=404)
+
+    del tasks_db[task_id]
+    return JSONResponse({"message": f"Task {task_id} deleted"})
+
+
+routes = [
+    Route("/", homepage),
+    Route("/health", health_check),
+    Route("/tasks", list_tasks, methods=["GET"]),
+    Route("/tasks", create_task, methods=["POST"]),
+    Route("/tasks/{task_id:int}", get_task, methods=["GET"]),
+    Route("/tasks/{task_id:int}", update_task, methods=["PUT"]),
+    Route("/tasks/{task_id:int}", delete_task, methods=["DELETE"]),
+]
+
+app = Starlette(debug=True, routes=routes)
+
+
+def create_app(debug: bool = False) -> Starlette:
+    return Starlette(debug=debug, routes=routes)
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8003)