feat: basic trophy api

main.py

@@ -2,7 +2,7 @@
 from fastapi import FastAPI, Depends
 from bson.objectid import ObjectId
 from typing import List
-from routers import notifications_router, users_router, activities_router, groups_router
+from routers import notifications_router, users_router, activities_router, groups_router, trophies_router
 from fastapi import FastAPI
 from database import close_mongo_connection, connect_to_mongo
 from fastapi.middleware.cors import CORSMiddleware
@@ -30,6 +30,7 @@
     notifications_router.router, prefix="/api/notification", tags=["notifications"]
 )
 app.include_router(groups_router.router, prefix="/api/group", tags=["groups"])
+app.include_router(trophies_router.router, prefix="/api/trophy", tags=["trophies"])
 
 @app.get("/api/cert")
 async def get_cert():

routers/notifications_router.py

@@ -38,6 +38,8 @@ async def get_notifications(user=Depends(get_current_user)):
     """
     Get Notifications
     """
+    if "admin" not in user["per"]:
+        raise HTTPException(status_code=403, detail="Permission denied")
     # Get notifications
     result = await db.zvms.notifications.find().to_list(1000)
     for i in result:

routers/trophies_router.py

@@ -0,0 +1,288 @@
+from typings.activity import ActivityMode
+from typings.trophy import (
+    Trophy,
+    TrophyAward,
+    TrophyMember,
+    TrophyMemberStatus,
+    TrophyStatus,
+)
+from bson import ObjectId
+from database import db
+from fastapi import HTTPException, APIRouter, Depends
+from util.get_class import get_classid_by_user_id
+from utils import compulsory_temporary_token, get_current_user, validate_object_id
+from datetime import datetime
+from pydantic import BaseModel
+
+router = APIRouter()
+
+
+@router.post("")
+async def create_trophy(request: Trophy, user=Depends(get_current_user)):
+    """
+    Create Trophy
+    """
+    if "admin" not in user["per"]:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    request.creator = user["id"]
+    if (
+        "secretary" in user["per"]
+        and "department" not in user["per"]
+        and "admin" not in user["per"]
+    ):
+        request.status = TrophyStatus.pending
+    elif "department" in user["per"] or "admin" in user["per"]:
+        request.status = TrophyStatus.effective
+    else:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    request.createdAt = datetime.now().isoformat()
+    # Create trophy
+    trophy = request.model_dump()
+    result = await db.zvms.trophies.insert_one(trophy)
+    return {
+        "status": "ok",
+        "code": 201,
+        "data": {"_id": str(result.inserted_id)},
+    }
+
+
+@router.get("")
+async def get_trophies(user=Depends(get_current_user)):
+    """
+    Get Trophies
+    """
+    # Get trophies
+    result = await db.zvms.trophies.find().to_list(1000)
+    for i in result:
+        i["_id"] = str(i["_id"])
+    return result
+
+
+class PutStatus(BaseModel):
+    status: TrophyStatus
+
+
+@router.put("/{trophy_oid}/status")
+async def update_trophy_status(
+    trophy_oid: str, request: PutStatus, user=Depends(get_current_user)
+):
+    """
+    Update Trophy Status
+    """
+    if "admin" not in user["per"] or "department" not in user["per"]:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    # Update trophy status
+    await db.zvms.trophies.update_one(
+        {"_id": validate_object_id(trophy_oid)},
+        {"$set": {"status": request.status}},
+    )
+    return {"status": "ok", "code": 200}
+
+
+class PutName(BaseModel):
+    name: str
+
+
+@router.put("/{trophy_oid}/name")
+async def update_trophy_name(
+    trophy_oid: str, request: PutName, user=Depends(get_current_user)
+):
+    """
+    Update Trophy Name
+    """
+    if "admin" not in user["per"]:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    # Update trophy name
+    await db.zvms.trophies.update_one(
+        {"_id": validate_object_id(trophy_oid)},
+        {"$set": {"name": request.name}},
+    )
+    return {"status": "ok", "code": 200}
+
+
+@router.delete("/{trophy_oid}")
+async def delete_trophy(trophy_oid: str, user=Depends(compulsory_temporary_token)):
+    """
+    Delete Trophy
+    """
+    trophy = await db.zvms.trophies.find_one({"_id": validate_object_id(trophy_oid)})
+    if "admin" not in user["per"] and user["id"] != trophy["creator"]:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    # Delete trophy
+    await db.zvms.trophies.delete_one({"_id": validate_object_id(trophy_oid)})
+    return {"status": "ok", "code": 200}
+
+
+@router.post("/{trophy_oid}/members")
+async def add_trophy_member(
+    trophy_oid: str, member: TrophyMember, user=Depends(get_current_user)
+):
+    """
+    Add Trophy Member
+    """
+    activity = await db.zvms.trophies.find_one({"_id": validate_object_id(trophy_oid)})
+
+    target = member.id
+
+    target_classid = get_classid_by_user_id(target)
+    user_classid = get_classid_by_user_id(user["id"])
+
+    if "admin" in user["per"] and "department" in user["per"]:
+        member.status = TrophyMemberStatus.pending
+        # The approval of the trophy needs to be approved by the member of department from the instructor
+    elif "secretary" in user["per"] and user_classid == target_classid:
+        member.status = TrophyMemberStatus.pending
+    elif target == user["id"]:
+        member.status = TrophyMemberStatus.pending
+    else:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    diction = member.model_dump()
+    diction["_id"] = diction["id"]
+    del diction["id"]
+    # Add trophy member
+    await db.zvms.trophies.update_one(
+        {"_id": validate_object_id(trophy_oid)},
+        {"$push": {"members": diction}},
+    )
+    return {"status": "ok", "code": 201}
+
+
+class PutMemberStatus(BaseModel):
+    status: TrophyMemberStatus
+
+
+@router.put("/{trophy_oid}/members/{member_oid}/status")
+async def update_trophy_member_status(
+    trophy_oid: str,
+    member_oid: str,
+    request: PutMemberStatus,
+    user=Depends(get_current_user),
+):
+    """
+    Update Trophy Member Status
+    """
+    trophy = await db.zvms.trophies.find_one({"_id": validate_object_id(trophy_oid)})
+    if "admin" in user["per"]:
+        pass
+    elif "department" in user["per"] and user["id"] == trophy["create"]:
+        pass
+    else:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    # Update trophy member status
+    await db.zvms.trophies.update_one(
+        {"_id": validate_object_id(trophy_oid), "members._id": member_oid},
+        {"$set": {"members.$.status": request.status}},
+    )
+    return {"status": "ok", "code": 200}
+
+
+class PutTrophyMemberMode(BaseModel):
+    mode: ActivityMode
+
+
+@router.put("/{trophy_oid}/members/{member_oid}/mode")
+async def update_trophy_member_mode(
+    trophy_oid: str,
+    member_oid: str,
+    request: PutTrophyMemberMode,
+    user=Depends(get_current_user),
+):
+    """
+    Update Trophy Member Mode
+    """
+    if user["id"] != member_oid and "admin" not in user["per"]:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    # Update trophy member mode
+    await db.zvms.trophies.update_one(
+        {"_id": validate_object_id(trophy_oid), "members._id": member_oid},
+        {"$set": {"members.$.mode": request.mode}},
+    )
+    return {"status": "ok", "code": 200}
+
+
+@router.delete("/{trophy_oid}/members/{member_oid}")
+async def delete_trophy_member(
+    trophy_oid: str, member_oid: str, user=Depends(compulsory_temporary_token)
+):
+    """
+    Delete Trophy Member
+    """
+    trophy = await db.zvms.trophies.find_one({"_id": validate_object_id(trophy_oid)})
+    target_classid = get_classid_by_user_id(member_oid)
+    user_classid = get_classid_by_user_id(user["id"])
+    if (
+        "admin" not in user["per"]
+        and not ("department" in user["per"] and user["id"] == trophy["creator"])
+        and member_oid != user["id"]
+        and not ("secretary" in user["per"] and user_classid == target_classid)
+    ):
+        raise HTTPException(status_code=403, detail="Permission denied")
+    # Delete trophy member
+    await db.zvms.trophies.update_one(
+        {"_id": validate_object_id(trophy_oid)},
+        {"$pull": {"members": {"_id": member_oid}}},
+    )
+    return {"status": "ok", "code": 200}
+
+
+@router.post("/{trophy_oid}/awards")
+async def add_trophy_award(
+    trophy_oid: str, award: TrophyAward, user=Depends(get_current_user)
+):
+    """
+    Add Trophy Award
+    """
+    trophy = await db.zvms.trophies.find_one({"_id": validate_object_id(trophy_oid)})
+    if "admin" not in user["per"] and user["id"] != trophy["creator"]:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    # Add trophy award
+    await db.zvms.trophies.update_one(
+        {"_id": validate_object_id(trophy_oid)},
+        {"$push": {"awards": award}},
+    )
+    return {"status": "ok", "code": 201}
+
+
+@router.delete("/{trophy_oid}/awards/{award_oid}")
+async def delete_trophy_award(
+    trophy_oid: str, award_oid: str, user=Depends(compulsory_temporary_token)
+):
+    """
+    Delete Trophy Award
+    """
+    trophy = await db.zvms.trophies.find_one({"_id": validate_object_id(trophy_oid)})
+    if "admin" not in user["per"] and user["id"] != trophy["creator"]:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    # Delete trophy award
+    await db.zvms.trophies.update_one(
+        {"_id": validate_object_id(trophy_oid)},
+        {
+            "$pull": {"awards": {"_id": award_oid}},
+        },
+    )
+    return {"status": "ok", "code": 200}
+
+
+class PutTrophyAwardDuration(BaseModel):
+    duration: float
+
+
+@router.put("/{trophy_oid}/awards/{award_oid}/duration")
+async def update_trophy_award_duration(
+    trophy_oid: str,
+    award_oid: str,
+    request: PutTrophyAwardDuration,
+    user=Depends(get_current_user),
+):
+    """
+    Update Trophy Award Duration
+    """
+    trophy = await db.zvms.trophies.find_one({"_id": validate_object_id(trophy_oid)})
+    if "admin" not in user["per"] and user["id"] != trophy["creator"]:
+        raise HTTPException(status_code=403, detail="Permission denied")
+    # Update trophy award duration
+    await db.zvms.trophies.update_one(
+        {"_id": validate_object_id(trophy_oid), "awards._id": award_oid},
+        {"$set": {"awards.$.duration": request.duration}},
+    )
+    return {"status": "ok", "code": 200}

typings/group.py

@@ -1,29 +1,29 @@
-from typing import Optional
-from pydantic import BaseModel, Field
-from enum import Enum
-from bson import ObjectId
-
-
-class GroupType(str, Enum):
-    class_ = "class"
-    permission = "permission"
-    group = "group"
-
-
-class UserPosition(str, Enum):
-    student = "student"
-    secretary = "secretary"
-    department = "department"
-    auditor = "auditor"
-    admin = "admin"
-    system = "system"
-
-
-
-class Group(BaseModel):
-    _id: ObjectId | str
-    name: str
-    type: GroupType
-    description: Optional[str]
-    permissions: list[UserPosition]
-    # Doesn't contain members, and it's not necessary to contain it in the model
+from typing import Optional
+from pydantic import BaseModel, Field
+from enum import Enum
+from bson import ObjectId
+
+
+class GroupType(str, Enum):
+    class_ = "class"
+    permission = "permission"
+    group = "group"
+
+
+class UserPosition(str, Enum):
+    student = "student"
+    secretary = "secretary"
+    department = "department"
+    auditor = "auditor"
+    admin = "admin"
+    system = "system"
+
+
+
+class Group(BaseModel):
+    _id: str
+    name: str
+    type: GroupType
+    description: Optional[str]
+    permissions: list[UserPosition]
+    # Doesn't contain members, and it's not necessary to contain it in the model