Feature: Site User endpoints

src/api/urls.py

@@ -8,6 +8,7 @@
 import image_asset.urls
 import nominees.urls
 import officers.urls
+import users.urls
 from api.auth import require_trusted_origin
 
 router = APIRouter(
@@ -22,3 +23,4 @@
 router.include_router(event.urls.router)
 router.include_router(honorary.urls.router)
 router.include_router(image_asset.urls.router)
+router.include_router(users.urls.router)

src/auth/crud.py

@@ -36,7 +36,12 @@ async def create_user_session(db_session: AsyncSession, session_id: str, computi
     )
     # ...or just update their "last_logged_in" time
     user_query = user_query.on_conflict_do_update(
-        index_elements=[SiteUserDB.computing_id], set_={"last_logged_in": now}
+        index_elements=[SiteUserDB.computing_id],
+        set_={
+            "last_logged_in": now,
+            # If the user's first_logged_in is null, set it to now, otherwise leave it as is.
+            "first_logged_in": sqlalchemy.func.coalesce(SiteUserDB.first_logged_in, now),
+        },
     )
     await db_session.execute(user_query)
 

src/dependencies.py

@@ -5,8 +5,8 @@
 import auth
 import auth.crud
 import database
-from auth.constants import COOKIE_SESSION_KEY
-from utils.permissions import is_user_election_admin, is_user_website_admin
+from auth.constants import COOKIE_SESSION_KEY, UserRole
+from utils.permissions import is_user_election_admin, is_user_role, is_user_website_admin
 
 # Dependency to ensure years in paths are valid
 # Honestly don't know if this DB will be running past year 3000
@@ -83,7 +83,7 @@ async def perm_election(db_session: database.DBSession, computing_id: LoggedInUs
 
 
 async def perm_admin(db_session: database.DBSession, computing_id: LoggedInUser):
-    if not await is_user_website_admin(computing_id, db_session):
+    if not await is_user_role(db_session, computing_id, UserRole.ADMIN):
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="must be an admin")
 
     return computing_id

src/load_test_db.py

@@ -31,6 +31,7 @@
 )
 from officers.tables import OfficerInfoDB, OfficerTermDB
 from translink.tables import TransLinkStaticScheduleDB
+from users.crud import UserRole, create_site_user, create_user_roles
 
 
 async def reset_db(engine):
@@ -231,7 +232,8 @@ async def load_test_officers_data(db_session: AsyncSession):
     await db_session.commit()
 
 
-SYSADMIN_COMPUTING_ID = "pkn4"
+SYSADMIN_COMPUTING_ID = "sysadmin"
+SYSADMIN_FULL_NAME = "System Administrator"
 
 
 async def load_sysadmin(db_session: AsyncSession):
@@ -241,7 +243,7 @@ async def load_sysadmin(db_session: AsyncSession):
     await create_new_officer_info(
         db_session,
         OfficerInfoDB(
-            legal_name="Puneet North",
+            legal_name=SYSADMIN_FULL_NAME,
             discord_id=None,
             discord_name=None,
             discord_nickname=None,
@@ -300,10 +302,20 @@ async def load_sysadmin(db_session: AsyncSession):
             photo_url=None,
         ),
     )
+    user_roles = [
+        SiteUserRoleDB(
+            computing_id=SYSADMIN_COMPUTING_ID,
+            role=UserRole.ADMIN,
+            added_by=SYSADMIN_COMPUTING_ID,
+            created_at=datetime.now(UTC),
+        )
+    ]
+    create_user_roles(db_session, user_roles)
     await db_session.commit()
 
 
-WEBMASTER_COMPUTING_ID = "jbriones"
+WEBMASTER_COMPUTING_ID = "webmaster"
+WEBMASTER_FULL_NAME = "Web Master"
 
 
 async def load_webmaster(db_session: AsyncSession):
@@ -313,7 +325,7 @@ async def load_webmaster(db_session: AsyncSession):
     await create_new_officer_info(
         db_session,
         OfficerInfoDB(
-            legal_name="Jon Andre Briones",
+            legal_name=WEBMASTER_FULL_NAME,
             discord_id=None,
             discord_name=None,
             discord_nickname=None,
@@ -330,7 +342,7 @@ async def load_webmaster(db_session: AsyncSession):
             position=OfficerPositionEnum.FIRST_YEAR_REPRESENTATIVE,
             start_date=date.today() - timedelta(days=(365 * 3)),
             end_date=date.today() - timedelta(days=(365 * 2)),
-            nickname="Jon Andre Briones",
+            nickname=WEBMASTER_FULL_NAME,
             favourite_course_0="CMPT 379",
             favourite_course_1="CMPT 371",
             favourite_pl_0="TypeScript",
@@ -355,6 +367,15 @@ async def load_webmaster(db_session: AsyncSession):
             photo_url=None,
         ),
     )
+    user_roles = [
+        SiteUserRoleDB(
+            computing_id=WEBMASTER_COMPUTING_ID,
+            role=UserRole.ADMIN,
+            added_by=SYSADMIN_COMPUTING_ID,
+            created_at=datetime.now(UTC),
+        )
+    ]
+    create_user_roles(db_session, user_roles)
     await db_session.commit()
 
 
@@ -413,8 +434,8 @@ async def load_test_elections_data(db_session: AsyncSession):
     await create_nominee_info(
         db_session,
         NomineeInfoDB(
-            computing_id="pkn4",
-            full_name="Puneet North",
+            computing_id="sysadmin",
+            full_name=SYSADMIN_FULL_NAME,
             linked_in="linkedin.com/john-doe3",
             instagram="john_doe 3",
             email="john_do3e@doe.com",

src/users/crud.py

@@ -0,0 +1,50 @@
+from collections.abc import Sequence
+
+from sqlalchemy import delete, select
+from sqlalchemy.orm import selectinload
+
+import database
+from auth.constants import UserRole
+from auth.tables import SiteUserDB, SiteUserRoleDB
+
+
+def create_site_user(db_session: database.DBSession, user: SiteUserDB) -> None:
+    db_session.add(user)
+
+
+async def get_all_users(db_session: database.DBSession) -> Sequence[SiteUserDB]:
+    query = select(SiteUserDB).options(selectinload(SiteUserDB.roles)).order_by(SiteUserDB.last_logged_in.desc())
+
+    return (await db_session.scalars(query)).all()
+
+
+def create_user_roles(db_session: database.DBSession, roles: list[SiteUserRoleDB]) -> None:
+    db_session.add_all(roles)
+
+
+async def get_user_for_role_update(db_session: database.DBSession, computing_id: str) -> SiteUserDB | None:
+    query = (
+        select(SiteUserDB)
+        .where(SiteUserDB.computing_id == computing_id)
+        .options(selectinload(SiteUserDB.roles))
+        .with_for_update()
+    )
+    user = await db_session.scalar(query)
+
+    return user
+
+
+async def delete_user_roles(db_session: database.DBSession, computing_id: str, roles_to_remove: set[UserRole]) -> None:
+    query = (
+        delete(SiteUserRoleDB)
+        .where(SiteUserRoleDB.computing_id == computing_id, SiteUserRoleDB.role.in_(roles_to_remove))
+        .execution_options(synchronize_session="fetch")
+    )
+
+    await db_session.execute(query)
+
+
+async def delete_user(db_session: database.DBSession, computing_id: str) -> None:
+    query = delete(SiteUserDB).where(SiteUserDB.computing_id == computing_id)
+
+    await db_session.execute(query)

src/users/models.py

@@ -0,0 +1,35 @@
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from auth.constants import UserRole
+from constants import COMPUTING_ID_LEN
+
+
+class SiteUserBase(BaseModel):
+    computing_id: str = Field(
+        ...,
+        min_length=1,
+        max_length=COMPUTING_ID_LEN,
+        description="Computing ID of the user",
+    )
+
+    roles: list[UserRole] = Field(
+        [],
+        description="List of roles assigned to the user",
+    )
+
+    @field_validator("roles")
+    @classmethod
+    def roles_must_be_unique(cls, roles: list[UserRole]) -> list[UserRole]:
+        if len(roles) != len(set(roles)):
+            raise ValueError("Roles must be unique")
+        return roles
+
+
+class SiteUserCreate(SiteUserBase):
+    pass
+
+
+class SiteUserUpdate(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    roles: list[UserRole] = Field(..., description="Complete set of roles assigned to the user")