Add invitation endpoints

src/api/app.py

@@ -14,6 +14,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 .invitations import router as invitations_router
 from .sessions import router as sessions_router
 from .users import router as users_router
 
@@ -59,6 +60,7 @@ app.include_router(users_router)
 app.include_router(sessions_router)
 app.include_router(categories_router)
 app.include_router(events_router)
+app.include_router(invitations_router)
 
 
 @app.get("/ping")

src/api/invitations.py

@@ -0,0 +1,228 @@
+from datetime import UTC, datetime
+from typing import Literal, cast, final
+
+from beanie import Link, PydanticObjectId
+from fastapi import APIRouter, HTTPException, Response, status
+from pydantic import BaseModel
+
+from src.db.models.event import Event
+from src.db.models.invitation import Invitation
+from src.db.models.user import User
+from src.utils.db import IdNotFoundError, MissingIdError, UnfetchedLinkError, expr
+from src.utils.logging import get_logger
+
+from .auth import CurrentUserDep
+
+__all__ = ["router"]
+
+log = get_logger(__name__)
+
+base_router = APIRouter(tags=["Invitations"])
+invitations_router = APIRouter(tags=["Invitations"], prefix="/invitations")
+
+
+class _BaseInvitationData(BaseModel):
+    """Base class for all invitation data classes."""
+
+    event_id: PydanticObjectId
+    invitee_id: PydanticObjectId
+
+
+@final
+class InvitationData(_BaseInvitationData):
+    """Information about an invitation sent to the user."""
+
+    id: PydanticObjectId
+    invitor_id: PydanticObjectId
+    status: Literal["accepted", "declined", "pending"]
+    responded_at: datetime | None
+    sent_at: datetime
+
+    @classmethod
+    def from_invitation(cls, invitation: Invitation) -> "InvitationData":
+        """Construct InvitationData from database Invitation object."""
+        if invitation.id is None:
+            raise MissingIdError(invitation)
+
+        if isinstance(invitation.invitee, Link):
+            raise UnfetchedLinkError(invitation.invitee)
+        if invitation.invitee.id is None:
+            raise MissingIdError(invitation.invitee)
+
+        if isinstance(invitation.event, Link):
+            raise UnfetchedLinkError(invitation.event)
+        if invitation.event.id is None:
+            raise MissingIdError(invitation.event)
+
+        if isinstance(invitation.invitor, Link):
+            raise UnfetchedLinkError(invitation.invitor)
+        if invitation.invitor.id is None:
+            raise MissingIdError(invitation.invitor)
+
+        return cls(
+            id=invitation.id,
+            event_id=invitation.event.id,
+            invitee_id=invitation.invitee.id,
+            invitor_id=invitation.invitor.id,
+            status=invitation.status,
+            sent_at=invitation.sent_at,
+            responded_at=invitation.responded_at,
+        )
+
+
+@final
+class InvitationCreateData(_BaseInvitationData):
+    """Data necessary to create an invitation.
+
+    This structure is intended to be used for POST requests.
+    """
+
+    async def create_invitation(self, user: User) -> Invitation:
+        """Create a new invitation in the database.
+
+        If one the event or invitee doesn't exist, the IdNotFoundError will be raised.
+        """
+        event = await Event.get(self.event_id)
+        if event is None:
+            raise IdNotFoundError(self.event_id, Event)
+
+        invitee = await User.get(self.invitee_id)
+        if invitee is None:
+            raise IdNotFoundError(self.invitee_id, User)
+
+        invitation = Invitation(
+            event=event,
+            invitee=invitee,
+            invitor=user,
+            status="pending",
+        )
+        return await invitation.create()
+
+
+@base_router.get("/users/{user_id}/invitations")
+async def get_user_invitatinos(user_id: PydanticObjectId, user: CurrentUserDep) -> list[InvitationData]:
+    """Get all invitations for a given user.
+
+    Note that this endpoint only allows you to access the invitations you created.
+    """
+    if user.id is None:
+        raise MissingIdError(user)
+
+    if user.id != user_id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You can only access your own invitations.")
+
+    invitations = await Invitation.find(expr(Invitation.invitor).id == user_id, fetch_links=True).to_list()
+    return [InvitationData.from_invitation(invitation) for invitation in invitations]
+
+
+@base_router.get("/users/{user_id}/invitations/incomming")
+async def get_user_incomming_invitatinos(user_id: PydanticObjectId, user: CurrentUserDep) -> list[InvitationData]:
+    """Get all incoming invitations for a given user.
+
+    Note that this endpoint only allows you to access the invitations you received.
+    """
+    if user.id is None:
+        raise MissingIdError(user)
+
+    if user.id != user_id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You can only access your own invitations.")
+
+    invitations = await Invitation.find(expr(Invitation.invitee).id == user_id, fetch_links=True).to_list()
+
+    return [InvitationData.from_invitation(invitation) for invitation in invitations]
+
+
+@invitations_router.get("{invitation_id}")
+async def get_invitation(invitation_id: PydanticObjectId, user: CurrentUserDep) -> InvitationData:
+    """Get information about a specific invitation.
+
+    You can only access the invitations you received or created.
+    """
+    invitation = await Invitation.get(invitation_id, fetch_links=True)
+
+    if invitation is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invitation not found.")
+
+    if cast(User, invitation.invitee).id != user.id and cast(User, invitation.invitor).id != user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only access invitations you received or created.",
+        )
+
+    return InvitationData.from_invitation(invitation)
+
+
+@invitations_router.post("")
+async def create_invitation(data: InvitationCreateData, user: CurrentUserDep) -> InvitationData:
+    """Create a new invitation."""
+    invitation = await data.create_invitation(user)
+    return InvitationData.from_invitation(invitation)
+
+
+@invitations_router.delete("{invitation_id}")
+async def delete_invitation(invitation_id: PydanticObjectId, user: CurrentUserDep) -> Response:
+    """Delete an invitation."""
+    invitation = await Invitation.get(invitation_id, fetch_links=True)
+
+    if invitation is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invitation not found.")
+
+    if cast(User, invitation.invitor).id != user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You can only delete your invitations you created.",
+        )
+
+    _ = await invitation.delete()
+    return Response(status.HTTP_204_NO_CONTENT)
+
+
+@invitations_router.post("{invitation_id}/accept")
+async def accept_invitation(invitation_id: PydanticObjectId, user: CurrentUserDep) -> Response:
+    """Accept an invitation."""
+    invitation = await Invitation.get(invitation_id, fetch_links=True)
+    if invitation is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invitation not found.")
+
+    if cast(User, invitation.invitee).id != user.id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You can only accept your own invitations.")
+
+    if invitation.status != "pending":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="You can only accept pending invitations.",
+        )
+
+    invitation.status = "accepted"
+    invitation.responded_at = datetime.now(tz=UTC)
+    _ = await invitation.save()
+
+    return Response(status.HTTP_204_NO_CONTENT)
+
+
+@invitations_router.post("{invitation_id}/decline")
+async def decline_invitation(invitation_id: PydanticObjectId, user: CurrentUserDep) -> Response:
+    """Decline an invitation."""
+    invitation = await Invitation.get(invitation_id, fetch_links=True)
+    if invitation is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invitation not found.")
+
+    if cast(User, invitation.invitee).id != user.id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You can only decline your own invitations.")
+
+    if invitation.status != "pending":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="You can only decline pending invitations.",
+        )
+
+    invitation.status = "declined"
+    invitation.responded_at = datetime.now(tz=UTC)
+    _ = await invitation.save()
+
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+router = APIRouter()
+router.include_router(base_router)
+router.include_router(invitations_router)

src/db/models/invitation.py

@@ -15,6 +15,7 @@ class Invitation(Document):
 
     event: Annotated[BeanieLink[Event], Indexed()]
     invitee: Annotated[BeanieLink[User], Indexed()]
+    invitor: Annotated[BeanieLink[User], Indexed()]
     status: Literal["accepted", "declined", "pending"] = "pending"
     sent_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     responded_at: datetime | None = None