Add events endpoints

src/api/app.py

@@ -13,6 +13,7 @@ from src.utils.logging import get_logger
 
 from .auth import router as auth_router
 from .categories import router as categories_router
+from .events import router as events_router
 from .sessions import router as sessions_router
 from .users import router as users_router
 
@@ -57,6 +58,7 @@ app.include_router(auth_router)
 app.include_router(users_router)
 app.include_router(sessions_router)
 app.include_router(categories_router)
+app.include_router(events_router)
 
 
 @app.get("/ping")

src/api/events.py

@@ -0,0 +1,267 @@
+from datetime import date, datetime, time
+from typing import Annotated, final
+
+from beanie import PydanticObjectId
+from fastapi import APIRouter, Body, HTTPException, status
+from fastapi.responses import Response
+from pydantic import BaseModel, StringConstraints, field_validator
+from pydantic_extra_types.color import Color
+
+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.db import from_id_list, get_id_list, update_document
+from src.utils.logging import get_logger
+
+from .auth import CurrentUserDep
+
+__all__ = ["router"]
+
+log = get_logger(__name__)
+
+events_router = APIRouter(tags=["Events"], prefix="/events", dependencies=[LoggedInDep])
+base_router = APIRouter(tags=["Events"], dependencies=[LoggedInDep])
+
+
+class _BaseEventData(BaseModel):
+    """Base model holding data about an event."""
+
+    title: Annotated[str, StringConstraints(min_length=1, max_length=50)]
+    description: Annotated[str, StringConstraints(max_length=1000)]
+    category_ids: list[PydanticObjectId]
+    start_date: date
+    start_time: time
+    end_date: date
+    end_time: time
+    color: Color
+    attendee_ids: list[PydanticObjectId]
+
+    @field_validator("color", mode="after")
+    @classmethod
+    def validate_color(cls, value: Color) -> Color:
+        """Validate the color."""
+        if len(value.as_rgb_tuple()) == 4:
+            raise ValueError("Alpha channel is not allowed in colors")
+        return value
+
+
+@final
+class EventData(_BaseEventData):
+    """Data about an event sent to the user."""
+
+    owner_user_id: PydanticObjectId
+    created_at: datetime
+
+    @classmethod
+    def from_event(cls, event: Event) -> "EventData":
+        if event.user.id is None:
+            raise ValueError("Got an event without owner user ID")
+
+        return cls(
+            owner_user_id=event.user.id,
+            title=event.title,
+            description=event.description,
+            category_ids=get_id_list(event.categories),
+            start_date=event.start_date,
+            start_time=event.start_time,
+            end_date=event.end_date,
+            end_time=event.end_time,
+            color=Color(event.color),
+            attendee_ids=get_id_list(event.attendees),
+            created_at=event.created_at,
+        )
+
+
+@final
+class EventCreateData(_BaseEventData):
+    """Data necessary to create a new event.
+
+    This structure is intended to be used for POST & PUT requests.
+    """
+
+    async def create_event(self, user: User) -> Event:
+        """Create a new event in the database.
+
+        If one of the attendees or categories is not found, IdNotFoundError will be raised.
+        """
+        event = Event(
+            user=user,
+            title=self.title,
+            description=self.description,
+            categories=await from_id_list(self.category_ids, Category),
+            start_date=self.start_date,
+            start_time=self.start_time,
+            end_date=self.end_date,
+            end_time=self.end_time,
+            color=self.color.as_hex(format="long"),
+            attendees=await from_id_list(self.attendee_ids, User),
+        )
+        return await event.create()
+
+    async def update_event(self, event: Event) -> Event:
+        """Update an existing event, overwriting all relevant data."""
+        return await update_document(
+            event,
+            title=self.title,
+            description=self.description,
+            categories=await from_id_list(self.category_ids, Category),
+            start_date=self.start_date,
+            start_time=self.start_time,
+            end_date=self.end_date,
+            end_time=self.end_time,
+            color=self.color.as_hex(format="long"),
+            attendees=await from_id_list(self.attendee_ids, User),
+        )
+
+
+@final
+class PartialEventUpdateData(_BaseEventData):
+    """Data necessary to perform a partial update of the event."""
+
+    title: Annotated[str, StringConstraints(min_length=1, max_length=50)] | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
+    description: Annotated[str, StringConstraints(max_length=1000)] | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
+    category_ids: list[PydanticObjectId] | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
+    start_date: date | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
+    start_time: time | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
+    end_date: date | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
+    end_time: time | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
+    color: Color | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
+    attendee_ids: list[PydanticObjectId] | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
+
+    async def update_event(self, event: Event) -> Event:
+        """Update an existing event, overwriting it with data that were specified."""
+        update_dct = {}
+
+        # TODO: This repetitive code could be simplified
+        if self.title is not None:
+            update_dct["title"] = self.title
+        if self.description is not None:
+            update_dct["description"] = self.description
+        if self.category_ids is not None:
+            update_dct["categories"] = await from_id_list(self.category_ids, Category)
+        if self.start_date is not None:
+            update_dct["start_date"] = self.start_date
+        if self.start_time is not None:
+            update_dct["start_time"] = self.start_time
+        if self.end_date is not None:
+            update_dct["end_date"] = self.end_date
+        if self.end_time is not None:
+            update_dct["end_time"] = self.end_time
+        if self.color is not None:
+            update_dct["color"] = self.color.as_hex(format="long")
+        if self.attendee_ids is not None:
+            update_dct["attendees"] = await from_id_list(self.attendee_ids, User)
+
+        return await update_document(event, **update_dct)
+
+
+@base_router.get("/users/{user_id}/events")
+async def get_user_events(user_id: PydanticObjectId, user: CurrentUserDep) -> list[EventData]:
+    """Get all events 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",
+        )
+
+    events = await (Event.find(Event.user == user, fetch_links=True)).to_list()
+    return [EventData.from_event(event) for event in events]
+
+
+@events_router.get("{event_id}")
+async def get_event(event_id: PydanticObjectId, user: CurrentUserDep) -> EventData:
+    """Get an event by ID."""
+    event = await Event.get(event_id)
+    if event is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event with given id doesn't exist")
+    if event.owner != user:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access your own events",
+        )
+
+    return EventData.from_event(event)
+
+
+@events_router.delete("{event_id}")
+async def delete_event(event_id: PydanticObjectId, user: CurrentUserDep) -> Response:
+    """Delete an event by ID."""
+    event = await Event.get(event_id)
+    if event is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event with given id doesn't exist")
+    if event.owner != user:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access your own events",
+        )
+
+    # TODO: First remove any associations to this event
+
+    _ = await event.delete()
+
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+@events_router.post("")
+async def create_event(
+    user: CurrentUserDep,
+    event_data: Annotated[EventCreateData, Body()],
+) -> EventData:
+    """Create a new event."""
+    event = await event_data.create_event(user)
+    return EventData.from_event(event)
+
+
+@events_router.put("{event_id}")
+async def overwrite_event(
+    event_id: PydanticObjectId,
+    user: CurrentUserDep,
+    event_data: Annotated[EventCreateData, Body()],
+) -> EventData:
+    """Overwrite a specific event."""
+    event = await Event.get(event_id)
+    if event is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event with given id doesn't exist")
+    if event.owner != user:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access your own events",
+        )
+
+    event = await event_data.update_event(event)
+    return EventData.from_event(event)
+
+
+@events_router.patch("/{event_id}")
+async def partial_update_event(
+    event_id: PydanticObjectId,
+    user: CurrentUserDep,
+    partial_data: Annotated[PartialEventUpdateData, Body()],
+) -> EventData:
+    """Partially update an event."""
+    # 1. Fetch the event
+    event = await Event.get(event_id)
+    if event is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Event with given id doesn't exist",
+        )
+
+    # 2. Check ownership
+    if event.owner != user:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access your own events",
+        )
+
+    # 3. Update the event with the partial data
+    updated_event = await partial_data.update_event(event)
+
+    # 4. Return the updated data
+    return EventData.from_event(updated_event)
+
+
+router = APIRouter()
+router.include_router(events_router)
+router.include_router(base_router)