Draft: example - add task routes

example/app/main.py

@@ -1,8 +1,8 @@
-from fastapi import FastAPI
+from app.routes.tasks import task_routes
 from app.routes.users import user_routes
+from fastapi import FastAPI
 
-
-app = FastAPI(title="Example FastAPI REST application", routes=[*user_routes])
+app = FastAPI(title="Example FastAPI REST application", routes=[*user_routes, *task_routes])
 
 
 @app.get("/")

example/app/models.py

@@ -1,9 +1,11 @@
 import enum
-from uuid import UUID
 from datetime import datetime
+from uuid import UUID
+
 from pydantic import BaseModel
 
 
+# Users
 class Priority(str, enum.Enum):
     LOW = "LOW"
     MEDIUM = "MEDIUM"
@@ -26,9 +28,20 @@ class UserUpdate(UserCreate):
     pass
 
 
+# Tasks
 class Task(BaseModel):
     id: UUID
     user_id: UUID
     title: str
     priority: Priority
     created_at: datetime
+
+
+class TaskCreate(BaseModel):
+    user_id: UUID
+    title: str
+    priority: Priority
+
+
+class TaskUpdate(TaskCreate):
+    pass

example/app/routes/tasks.py

@@ -1,7 +1,109 @@
-from fastapi import APIRouter
-from fastapi.responses import JSONResponse
+import random
+from datetime import datetime
+from uuid import UUID, uuid4
+
+from app.data import tasks
+from app.models import Task, TaskCreate, TaskUpdate
 from fastapi.exceptions import HTTPException
+from fastapi.responses import JSONResponse
+from fastapi.routing import APIRoute
+
+
+class __TaskRoutes:
+    @staticmethod
+    async def get_random_task() -> Task:
+        rand_index = random.randint(0, len(tasks) - 1)
+        return Task(**tasks[rand_index])
+
+    @staticmethod
+    async def get_all_tasks() -> list[Task]:
+        return tasks
+
+    @staticmethod
+    async def get_task_by_id(task_id: UUID) -> Task:
+        task = next((t for t in tasks if t["id"] == task_id), None)
+        return Task(**task)
+
+    @staticmethod
+    async def create_task(new_task: TaskCreate) -> Task:
+        t = Task(
+            id=uuid4(),
+            user_id=new_task.user_id,
+            title=new_task.title,
+            priority=new_task.priority,
+            created_at=datetime.now(),
+        )
+        tasks.append(t.model_dump())
+
+        return t
+
+    @staticmethod
+    async def update_task(task_id: UUID, task: TaskUpdate) -> Task:
+        for i, task in enumerate(tasks):
+            if task["id"] == task_id:
+                task_updated = Task(**{**task["id"], **task.model_dump(), "created_at": datetime.now()})
+                tasks[i] = task_updated.model_dump()
+                return task_updated
+
+        raise HTTPException(status_code=404, detail="Task not found")
+
+    @staticmethod
+    async def delete_task(task_id: UUID) -> bool:
+        global tasks
+
+        if task_id not in [t["id"] for t in tasks]:
+            raise HTTPException(status_code=404, detail="Task not found")
+
+        tasks = [t for t in tasks if t["id"] != task_id]
+        return True
 
-from app.data import users, tasks
 
-tasks_router = APIRouter(prefix="/tasks")
+task_routes = [
+    APIRoute(
+        path="/tasks",
+        endpoint=__TaskRoutes.get_random_task,
+        methods=["GET"],
+        response_class=JSONResponse,
+        response_model=Task,
+        status_code=200,
+    ),
+    APIRoute(
+        path="/tasks/all",
+        endpoint=__TaskRoutes.get_all_tasks,
+        methods=["GET"],
+        response_class=JSONResponse,
+        response_model=list[Task],
+        status_code=200,
+    ),
+    APIRoute(
+        path="/tasks/{task_id:uuid}",
+        endpoint=__TaskRoutes.get_task_by_id,
+        methods=["GET"],
+        response_class=JSONResponse,
+        response_model=Task,
+        status_code=200,
+    ),
+    APIRoute(
+        path="/tasks/",
+        endpoint=__TaskRoutes.create_task,
+        methods=["POST"],
+        response_class=JSONResponse,
+        response_model=Task,
+        status_code=201,
+    ),
+    APIRoute(
+        path="/tasks/{task_id:uuid}",
+        endpoint=__TaskRoutes.update_task,
+        methods=["PATCH"],
+        response_class=JSONResponse,
+        response_model=Task,
+        status_code=200,
+    ),
+    APIRoute(
+        path="/tasks/{task_id:uuid}",
+        endpoint=__TaskRoutes.delete_task,
+        methods=["DELETE"],
+        response_class=JSONResponse,
+        status_code=200,
+    ),
+]

example/app/routes/users.py

@@ -1,14 +1,20 @@
-from uuid import UUID, uuid4
+import random
 from datetime import datetime
-from fastapi.routing import APIRoute
-from fastapi.responses import JSONResponse
-from fastapi.exceptions import HTTPException
+from uuid import UUID, uuid4
 
-from app.models import Users, UserCreate, UserUpdate
-from app.data import users
+from app.data import tasks, users
+from app.models import Task, UserCreate, Users, UserUpdate
+from fastapi.exceptions import HTTPException
+from fastapi.responses import JSONResponse
+from fastapi.routing import APIRoute
 
 
 class __UserRoutes:
+    @staticmethod
+    async def get_random_user() -> Users:
+        rand_index = random.randint(0, len(users) - 1)
+        return Users(**users[rand_index])
+
     @staticmethod
     async def get_all() -> list[Users]:
         return [Users(**user) for user in users]
@@ -53,14 +59,23 @@ async def delete(user_id: UUID) -> bool:
 
         return True
 
+    @staticmethod
+    async def get_tasks_for_user(user_id: UUID) -> list[Task]:
+        user = next((u for u in users if u["id"] == user_id), None)
+
+        if not user:
+            raise HTTPException(status_code=404, detail="User not found")
+
+        return [Task(**t) for t in tasks if t["user_id"] == user_id]
+
 
 user_routes = [
     APIRoute(
         path="/users",
-        endpoint=__UserRoutes.get_all,
+        endpoint=__UserRoutes.get_random_user,
         methods=["GET"],
         response_class=JSONResponse,
-        response_model=list[Users],
+        response_model=Users,
         status_code=200,
     ),
     APIRoute(
@@ -71,6 +86,14 @@ async def delete(user_id: UUID) -> bool:
         response_model=Users,
         status_code=200,
     ),
+    APIRoute(
+        path="/users/all",
+        endpoint=__UserRoutes.get_all,
+        methods=["GET"],
+        response_class=JSONResponse,
+        response_model=list[Users],
+        status_code=200,
+    ),
     APIRoute(
         path="/users/",
         endpoint=__UserRoutes.create,
@@ -80,7 +103,7 @@ async def delete(user_id: UUID) -> bool:
         status_code=201,
     ),
     APIRoute(
-        path="/users/",
+        path="/users/{user_id:uuid}",
         endpoint=__UserRoutes.update,
         methods=["PUT"],
         response_class=JSONResponse,
@@ -95,4 +118,12 @@ async def delete(user_id: UUID) -> bool:
         response_model=list[Users],
         status_code=200,
     ),
+    APIRoute(
+        path="/users/{user_id:uuid}/tasks",
+        endpoint=__UserRoutes.get_tasks_for_user,
+        methods=["GET"],
+        response_class=JSONResponse,
+        response_model=list[Task],
+        status_code=200,
+    ),
 ]