Type-hint Link db attributes more appropriately

src/api/auth/api.py

@@ -1,5 +1,6 @@
 from typing import Annotated, Literal
 
+from beanie import Link
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.responses import Response
 from fastapi.security import OAuth2PasswordRequestForm
@@ -89,8 +90,11 @@ async def refresh(cur_token: RefreshTokenDep) -> AccessTokenResponseBase:
 
     await invalidate_access_tokens(cur_db_token)
 
+    if isinstance(cur_db_token.user, Link):
+        raise TypeError("The user attribute must be fetched")
+
     if cur_db_token.user.id is None:
-        raise RuntimeError("Never")
+        raise ValueError("Got user without id")
 
     # We can now be certain there aren't any existing valid access tokens made from this refresh token.
     # Let's create a new access token.

src/api/auth/dependencies.py

@@ -1,5 +1,6 @@
 from typing import Annotated, Literal, TypeAlias
 
+from beanie import Link
 from fastapi import Depends, HTTPException, Security, status
 from fastapi.security import OAuth2PasswordBearer, SecurityScopes
 
@@ -83,6 +84,10 @@ async def get_current_user(
 ) -> User:
     """FastAPI dependency to get the currently logged in user from the JWT bearer access token."""
     _, db_token = await _get_access_token(token, security_scopes=security_scopes)
+
+    if isinstance(db_token.user, Link):
+        raise TypeError("The user attribute must be fetched")
+
     return db_token.user
 
 

src/api/sessions.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Literal, final
 
-from beanie import PydanticObjectId
+from beanie import Link, PydanticObjectId
 from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel
 
@@ -33,10 +33,18 @@ class UserSession(BaseModel):
     @classmethod
     def from_token(cls, token: Token) -> "UserSession":
         """Construct UserSession from database Token object."""
+        # This can only happen if the attendee wasn't yet committed to the db,
+        # within the context of this function, that should never happen.
         if token.id is None:
-            raise RuntimeError("Never")
-        if token.parent_token and token.parent_token.id is None:
-            raise RuntimeError("Never")
+            raise ValueError("Got token without id")
+
+        if token.parent_token:
+            # This function expects the parent token links to be fetched already
+            if isinstance(token.parent_token, Link):
+                raise TypeError("Parent token must be fetched before calling this function")
+
+            if token.parent_token.id is None:
+                raise ValueError("Got parent token without id")
 
         return cls(
             id=str(token.id),

src/db/models/category.py

@@ -11,7 +11,7 @@ from src.db.models.user import User
 class Category(Document):
     """Category table."""
 
-    user: Annotated[User, Annotated[Link[User], Indexed()]]
+    user: Annotated[User | Link[User], Annotated[Link[User], Indexed()]]
     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))

src/db/models/event.py

@@ -15,13 +15,13 @@ class Event(Document):
     user: Annotated[User, Annotated[Link[User], Indexed()]]
     title: str
     description: str
-    categories: Annotated[list[Category], Annotated[list[Link[Category]], Indexed()]]
+    categories: Annotated[list[Category | Link[Category]], Annotated[list[Link[Category]], Indexed()]]
     start_date: Annotated[date, Indexed()]
     start_time: time
     end_date: Annotated[date, Indexed()]
     end_time: time
     color: str  # TODO: Consider using a complex rgb type
-    attendees: Annotated[list[Link[User]], Indexed()]
+    attendees: Annotated[list[User | Link[User]], Annotated[list[Link[User]], Indexed()]]
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     @final

src/db/models/invitation.py

@@ -12,8 +12,8 @@ from src.db.models.user import User
 class Invitation(Document):
     """Invitation table."""
 
-    event: Annotated[Event, Annotated[Link[Event], Indexed()]]
-    invitee: Annotated[User, Annotated[Link[User], Indexed()]]
+    event: Annotated[Event | Link[User], Annotated[Link[Event], Indexed()]]
+    invitee: Annotated[User | Link[User], Annotated[Link[User], Indexed()]]
     status: Literal["accepted", "declined", "pending"] = "pending"
     sent_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     responded_at: datetime | None = None

src/db/models/notificaton.py

@@ -11,7 +11,7 @@ from src.db.models.user import User
 class Notification(Document):
     """Notification table."""
 
-    user: Annotated[User, Annotated[Link[User], Indexed()]]
+    user: Annotated[User | Link[User], Annotated[Link[User], Indexed()]]
     event_type: Annotated[Literal["reminder", "invitation"], Indexed()]
     message: str
     data: Any

src/db/models/token.py

@@ -12,9 +12,9 @@ from src.db.models.user import User
 class Token(Document):
     """Token table."""
 
-    user: Annotated[User, Annotated[Link[User], Indexed()]]
+    user: Annotated[User | Link[User], Annotated[Link[User], Indexed()]]
     tok_type: Literal["access", "refresh"]
-    parent_token: Annotated["Token | None", Annotated[Link["Token"] | None, Indexed()]] = None
+    parent_token: Annotated["Token | Link[Token] | None", Annotated[Link["Token"] | None, Indexed()]] = None
     revoked: bool = False
     issued_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     expires_at: datetime