Add honourary member endpoints

src/alembic/env.py

@@ -7,12 +7,13 @@
 
 import auth.tables
 import blog.tables
+import candidates.tables
 import database
 import elections.tables
+import event.tables
+import honorary.tables
 import nominees.tables
 import officers.tables
-import candidates.tables
-import event.tables
 import translink.tables
 from alembic import context
 

src/alembic/versions/3c00f28c26e2_create_honorary_member_table.py

@@ -0,0 +1,39 @@
+"""create honorary member table
+
+Revision ID: 3c00f28c26e2
+Revises: 42f855bec532
+Create Date: 2026-06-29 15:43:07.361827
+
+"""
+
+from collections.abc import Sequence
+from typing import Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "3c00f28c26e2"
+down_revision: str | None = "f0c99d0db277"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "honorary_member",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("name", sa.String(length=128), nullable=False),
+        sa.Column("start_date", sa.Date(), nullable=False),
+        sa.Column("end_date", sa.Date(), nullable=True),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_honorary_member")),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("honorary_member")
+    # ### end Alembic commands ###

src/honorary/constants.py

@@ -0,0 +1 @@
+HONORARY_MEMBER_MAX_LENGTH = 128

src/honorary/crud.py

@@ -0,0 +1,50 @@
+from datetime import date
+
+from sqlalchemy import delete, select
+
+import database
+from honorary.tables import HonoraryMemberDB
+
+
+async def get_all_honorary_members(db_session: database.DBSession) -> list[HonoraryMemberDB]:
+    query = select(HonoraryMemberDB).order_by(HonoraryMemberDB.start_date.desc())
+    return list((await db_session.scalars(query)).all())
+
+
+async def get_current_honorary_members(db_session: database.DBSession) -> list[HonoraryMemberDB]:
+    today = date.today()
+    query = (
+        select(HonoraryMemberDB)
+        .where(
+            (HonoraryMemberDB.start_date <= today)
+            & ((HonoraryMemberDB.end_date >= today) | HonoraryMemberDB.end_date.is_(None))
+        )
+        .order_by(HonoraryMemberDB.start_date.desc())
+    )
+    return list((await db_session.scalars(query)).all())
+
+
+async def get_honorary_member_by_id(
+    db_session: database.DBSession,
+    term_id: int,
+) -> HonoraryMemberDB | None:
+    return await db_session.scalar(select(HonoraryMemberDB).where(HonoraryMemberDB.id == term_id))
+
+
+def has_term_ended(honorary_member: HonoraryMemberDB) -> bool:
+    return honorary_member.end_date is not None and honorary_member.end_date < date.today()
+
+
+async def create_honorary_members(
+    db_session: database.DBSession,
+    honorary_members: list[HonoraryMemberDB],
+) -> list[HonoraryMemberDB]:
+    db_session.add_all(honorary_members)
+    return honorary_members
+
+
+async def delete_honorary_member(
+    db_session: database.DBSession,
+    honorary_member: HonoraryMemberDB,
+) -> None:
+    await db_session.execute(delete(HonoraryMemberDB).where(HonoraryMemberDB.id == honorary_member.id))

src/honorary/models.py

@@ -0,0 +1,29 @@
+from datetime import date
+
+from pydantic import BaseModel, ConfigDict, Field, computed_field
+
+from honorary.constants import HONORARY_MEMBER_MAX_LENGTH
+from utils import is_active_term
+
+
+class HonoraryMemberCreate(BaseModel):
+    name: str = Field(..., max_length=HONORARY_MEMBER_MAX_LENGTH)
+    start_date: date
+    end_date: date | None = None
+
+
+class HonoraryMember(HonoraryMemberCreate):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+
+    @computed_field
+    @property
+    def is_active(self) -> bool:
+        return is_active_term(start_date=self.start_date, end_date=self.end_date)
+
+
+class HonoraryMemberUpdate(BaseModel):
+    name: str | None = Field(None, max_length=HONORARY_MEMBER_MAX_LENGTH)
+    start_date: date | None = None
+    end_date: date | None = None

src/honorary/tables.py

@@ -0,0 +1,16 @@
+from datetime import date
+
+from sqlalchemy import Date, Integer, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from database import Base
+from honorary.constants import HONORARY_MEMBER_MAX_LENGTH
+
+
+class HonoraryMemberDB(Base):
+    __tablename__ = "honorary_member"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(HONORARY_MEMBER_MAX_LENGTH), nullable=False)
+    start_date: Mapped[date] = mapped_column(Date, nullable=False)
+    end_date: Mapped[date] = mapped_column(Date, nullable=True)

src/honorary/urls.py

@@ -0,0 +1,154 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.responses import JSONResponse
+
+import database
+import honorary.crud
+from dependencies import perm_admin
+from honorary.models import HonoraryMember, HonoraryMemberCreate, HonoraryMemberUpdate
+from honorary.tables import HonoraryMemberDB
+from utils.shared_models import DetailModel, SuccessResponse
+
+router = APIRouter(
+    prefix="/honorary",
+    tags=["honorary"],
+)
+
+
+@router.get(
+    "",
+    description="Get all honorary member terms",
+    response_model=list[HonoraryMember],
+    responses={403: {"description": "must be a website admin", "model": DetailModel}},
+    operation_id="get_all_honorary_members",
+    dependencies=[Depends(perm_admin)],
+)
+async def get_all_honorary_members(db_session: database.DBSession):
+    honorary_members = await honorary.crud.get_all_honorary_members(db_session)
+    return JSONResponse(
+        [
+            HonoraryMember.model_validate(member).model_dump(mode="json", exclude_unset=True)
+            for member in honorary_members
+        ]
+    )
+
+
+@router.get(
+    "/current",
+    description="Get all active honorary member terms",
+    response_model=list[HonoraryMember],
+    responses={403: {"description": "must be a website admin", "model": DetailModel}},
+    operation_id="get_current_honorary_members",
+    dependencies=[Depends(perm_admin)],
+)
+async def get_current_honorary_members(db_session: database.DBSession):
+    honorary_members = await honorary.crud.get_current_honorary_members(db_session)
+    return JSONResponse(
+        [
+            HonoraryMember.model_validate(member).model_dump(mode="json", exclude_unset=True)
+            for member in honorary_members
+        ]
+    )
+
+
+@router.post(
+    "",
+    description="Create honorary member terms",
+    response_model=list[HonoraryMember],
+    responses={
+        403: {"description": "must be a website admin", "model": DetailModel},
+        500: {"description": "failed to create honorary member terms", "model": DetailModel},
+    },
+    operation_id="create_honorary_members",
+    dependencies=[Depends(perm_admin)],
+)
+async def create_honorary_members(
+    db_session: database.DBSession,
+    body: list[HonoraryMemberCreate],
+):
+    new_members = [HonoraryMemberDB(**member.model_dump()) for member in body]
+    created_members = await honorary.crud.create_honorary_members(db_session, new_members)
+
+    await db_session.flush()
+
+    content = [
+        HonoraryMember.model_validate(member).model_dump(mode="json", exclude_unset=True) for member in created_members
+    ]
+
+    await db_session.commit()
+
+    return JSONResponse(content)
+
+
+@router.patch(
+    "/{term_id}",
+    description="Update an honorary member term. Ended terms cannot be updated.",
+    response_model=HonoraryMember,
+    responses={
+        403: {"description": "must be a website admin", "model": DetailModel},
+        404: {"description": "honorary member term does not exist", "model": DetailModel},
+        409: {"description": "honorary member term has already ended", "model": DetailModel},
+    },
+    operation_id="update_honorary_member",
+    dependencies=[Depends(perm_admin)],
+)
+async def update_honorary_member(
+    db_session: database.DBSession,
+    term_id: int,
+    body: HonoraryMemberUpdate,
+):
+    honorary_member = await honorary.crud.get_honorary_member_by_id(db_session, term_id)
+    if honorary_member is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"honorary member term with id={term_id} does not exist",
+        )
+
+    if honorary.crud.has_term_ended(honorary_member):
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="cannot update a term that has already ended",
+        )
+
+    update_data = body.model_dump(exclude_unset=True)
+    for key, value in update_data.items():
+        setattr(honorary_member, key, value)
+
+    await db_session.commit()
+    await db_session.refresh(honorary_member)
+
+    return JSONResponse(HonoraryMember.model_validate(honorary_member).model_dump(mode="json", exclude_unset=True))
+
+
+@router.delete(
+    "/{term_id}",
+    description="Delete an honorary member term. Ended terms cannot be deleted.",
+    response_model=SuccessResponse,
+    responses={
+        403: {"description": "must be a website admin", "model": DetailModel},
+        404: {"description": "honorary member term does not exist", "model": DetailModel},
+        409: {"description": "honorary member term has already ended", "model": DetailModel},
+    },
+    operation_id="delete_honorary_member",
+    dependencies=[Depends(perm_admin)],
+)
+async def delete_honorary_member(
+    db_session: database.DBSession,
+    term_id: int,
+):
+    honorary_member = await honorary.crud.get_honorary_member_by_id(db_session, term_id)
+    if honorary_member is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"honorary member term with id={term_id} does not exist",
+        )
+
+    if honorary.crud.has_term_ended(honorary_member):
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="cannot delete a term that has already ended",
+        )
+
+    await honorary.crud.delete_honorary_member(db_session, honorary_member)
+    await db_session.commit()
+
+    return SuccessResponse(success=True)

src/load_test_db.py

@@ -8,6 +8,8 @@
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
 
+import honorary.tables
+
 # NOTE: make sure you import from a file in your module which (at least) indirectly contains those
 # tables, or the current python context will not be able to find them & they won't be loaded
 from auth.crud import create_user_session, update_site_user

src/main.py

@@ -14,6 +14,7 @@
 import database
 import elections.urls
 import event.urls
+import honorary.urls
 import nominees.urls
 import officers.urls
 import permission.urls
@@ -78,6 +79,7 @@ async def lifespan(app: FastAPI):
 app.include_router(officers.urls.router)
 app.include_router(permission.urls.router)
 app.include_router(event.urls.router)
+app.include_router(honorary.urls.router)
 app.include_router(translink.urls.router)