Add endpoints to manage categories

src/api/app.py

@@ -12,6 +12,7 @@ from src.db.init import initialize_db
 from src.utils.logging import get_logger
 
 from .auth import router as auth_router
+from .categories import router as categories_router
 from .sessions import router as sessions_router
 from .users import router as users_router
 
@@ -55,6 +56,7 @@ app = FastAPI(
 app.include_router(auth_router)
 app.include_router(users_router)
 app.include_router(sessions_router)
+app.include_router(categories_router)
 
 
 @app.get("/ping")

src/api/categories.py

@@ -0,0 +1,213 @@
+from datetime import datetime
+from typing import Annotated, final
+
+from beanie import PydanticObjectId
+from fastapi import APIRouter, Body, HTTPException, Response, status
+from pydantic import BaseModel
+
+from src.api.auth.dependencies import LoggedInDep
+from src.db.models.category import Category
+from src.db.models.event import Event
+from src.db.models.user import User
+from src.utils.logging import get_logger
+
+from .auth import CurrentUserDep
+
+__all__ = ["CategoryData", "router"]
+
+log = get_logger(__name__)
+
+categories_router = APIRouter(tags=["Categories"], prefix="/categories", dependencies=[LoggedInDep])
+base_router = APIRouter(tags=["Categories"], dependencies=[LoggedInDep])
+
+
+@final
+class CategoryData(BaseModel):
+    """Data about a category sent to the user."""
+
+    owner_user_id: PydanticObjectId
+    name: str
+    color: str
+    created_at: datetime
+
+    @classmethod
+    def from_category(cls, category: Category) -> "CategoryData":
+        """Construct CategoryData from an instance of Category."""
+        if category.id is None:
+            raise ValueError("Got a category without id")
+
+        return cls(
+            owner_user_id=category.id,
+            name=category.name,
+            color=category.color,
+            created_at=category.created_at,
+        )
+
+
+@final
+class CategoryCreateData(BaseModel):
+    """Data necessary to create a new category.
+
+    This structure is intended to be used for POST & PUT requests.
+    """
+
+    name: str
+    color: str
+
+    async def create_category(self, user: User) -> Category:
+        """Create a new category in the database."""
+        cat = Category(user=user, name=self.name, color=self.color)
+        return await cat.create()
+
+    async def update_category(self, category: Category) -> Category:
+        """Update an existing category, overwriting all relevant data."""
+        updated = False
+
+        if category.name != self.name:
+            category.name = self.name
+            updated = True
+        if category.color != self.color:
+            category.color = self.color
+            updated = True
+
+        if updated:
+            category = await category.replace()
+        return category
+
+
+@final
+class PartialCategoryUpdateData(BaseModel):
+    """Data necessary to perform a partial update of the category.
+
+    This structure is intended to be used for PATCH requests.
+    """
+
+    name: str | None = None
+    color: str | None = None
+
+    async def update_category(self, category: Category) -> Category:
+        """Update an existing category, overwriting it with data that were specified."""
+        updated = False
+
+        if self.name and category.name != self.name:
+            category.name = self.name
+            updated = True
+        if self.color and category.color != self.color:
+            category.color = self.color
+            updated = True
+
+        if updated:
+            category = await category.replace()
+        return category
+
+
+@base_router.get("/users/{user_id}/categories")
+async def get_user_categories(user_id: PydanticObjectId, user: CurrentUserDep) -> list[CategoryData]:
+    """Get all categories that belong to given user."""
+    if user_id != user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access your own categories",
+        )
+
+    categories = await Category.find(Category.user == user).to_list()
+    return [CategoryData.from_category(category) for category in categories]
+
+
+@categories_router.get("{category_id}")
+async def get_category(category_id: PydanticObjectId, user: CurrentUserDep) -> CategoryData:
+    """Get a category by ID."""
+    category = await Category.get(category_id)
+    if category is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Category with given id doesn't exist")
+    if category.owner != user:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access your own categories",
+        )
+
+    return CategoryData.from_category(category)
+
+
+@categories_router.delete("{category_id}")
+async def delete_category(category_id: PydanticObjectId, user: CurrentUserDep) -> Response:
+    """Delete the given category.
+
+    If any events are associated with this category, this association will be removed.
+    """
+    category = await Category.get(category_id)
+    if category is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Category with given id doesn't exist")
+    if category.owner != user:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access your own categories",
+        )
+
+    # First remove the association to this category by events
+    # (this may leave the event without a category, however that's fine, as event category is optional
+    # and it's far more sensible solution than deleting all events with this category)
+    events = await Event.find(category in Event.categories).to_list()
+    for event in events:
+        event.categories.remove(category)
+    await Event.replace_many(events)
+
+    # We can now safely delete the category
+    _ = await category.delete()
+
+    return Response(status_code=status.HTTP_200_OK)
+
+
+@categories_router.post("")
+async def create_category(
+    user: CurrentUserDep,
+    category_data: Annotated[CategoryCreateData, Body()],
+) -> CategoryData:
+    """Create a new category."""
+    category = await category_data.create_category(user)
+    return CategoryData.from_category(category)
+
+
+@categories_router.put("{category_id}")
+async def overwrite_category(
+    category_id: PydanticObjectId,
+    user: CurrentUserDep,
+    category_data: Annotated[CategoryCreateData, Body()],
+) -> CategoryData:
+    """Overwrite a specific category."""
+    category = await Category.get(category_id)
+    if category is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Category with given id doesn't exist")
+    if category.owner != user:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access your own categories",
+        )
+
+    category = await category_data.update_category(category)
+    return CategoryData.from_category(category)
+
+
+@categories_router.patch("{category_id}")
+async def update_category(
+    category_id: PydanticObjectId,
+    user: CurrentUserDep,
+    category_data: Annotated[PartialCategoryUpdateData, Body()],
+) -> CategoryData:
+    """Overwrite a specific category."""
+    category = await Category.get(category_id)
+    if category is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Category with given id doesn't exist")
+    if category.owner != user:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access your own categories",
+        )
+
+    category = await category_data.update_category(category)
+    return CategoryData.from_category(category)
+
+
+router = APIRouter()
+router.include_router(categories_router)
+router.include_router(base_router)

src/db/models/category.py

@@ -12,7 +12,7 @@ class Category(Document):
     """Category table."""
 
     user: Annotated[User, Annotated[Link[User], Indexed()]]
-    name: str
+    name: str  # TODO: Should this be unique?
     color: str  # TODO: Consider using a complex rgb type
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))