121 lines
4.0 KiB
Python
121 lines
4.0 KiB
Python
from typing import Annotated
|
|
|
|
from asyncpg import Connection
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
|
|
import db.memberships as db
|
|
import settings.settings as settings
|
|
from api.models import User
|
|
from api.utils import get_current_user
|
|
from db.groups import (
|
|
check_group_author,
|
|
check_group_existence,
|
|
get_groupname_by_invite_code,
|
|
)
|
|
from db.internal import get_db_connection
|
|
from db.users import check_user_existence
|
|
|
|
memberships_router = APIRouter(prefix="/api/membership", tags=["memberships"])
|
|
|
|
|
|
@memberships_router.get("/me")
|
|
async def read_users_groups(
|
|
conn: Annotated[Connection, Depends(get_db_connection)],
|
|
current_user: Annotated[User, Depends(get_current_user)]
|
|
):
|
|
return await db.get_memberships_by_username(conn, current_user.username)
|
|
|
|
@memberships_router.post("/user")
|
|
async def read_users_any_memberships(
|
|
username: str,
|
|
conn: Annotated[Connection, Depends(get_db_connection)],
|
|
current_user: Annotated[User, Depends(get_current_user)]
|
|
):
|
|
if not await check_user_existence(conn, username):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User does not exist",
|
|
)
|
|
|
|
if not username == current_user.username and current_user.role not in settings.settings.admin_roles:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Not allowed",
|
|
)
|
|
|
|
return await db.get_memberships_by_username(conn, username)
|
|
|
|
@memberships_router.post("/group")
|
|
async def read_any_group_members(
|
|
groupname: str,
|
|
conn: Annotated[Connection, Depends(get_db_connection)],
|
|
current_user: Annotated[User, Depends(get_current_user)]
|
|
):
|
|
user_is_in_group = await db.check_membership_exists(conn, current_user.username, groupname)
|
|
if not user_is_in_group and current_user.role not in settings.settings.admin_roles:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Not allowed",
|
|
)
|
|
|
|
if not await check_group_existence(conn, groupname):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="No such group",
|
|
)
|
|
|
|
return await db.get_memberships_by_groupname(conn, groupname)
|
|
|
|
|
|
@memberships_router.post("/add")
|
|
async def add_membership(
|
|
username: str,
|
|
invite_code: str,
|
|
conn: Annotated[Connection, Depends(get_db_connection)],
|
|
current_user: Annotated[User, Depends(get_current_user)]
|
|
):
|
|
if username != current_user.username and current_user.role not in settings.settings.admin_roles:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Not allowed",
|
|
)
|
|
|
|
groupname = await get_groupname_by_invite_code(conn, invite_code)
|
|
if groupname is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Invite code is incorrect",
|
|
)
|
|
|
|
if not await check_user_existence(conn, username):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User does not exist",
|
|
)
|
|
|
|
if await db.check_membership_exists(conn, username, groupname):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="User is already a member",
|
|
)
|
|
|
|
return await db.create_membership(conn, username, groupname)
|
|
|
|
@memberships_router.post("/delete")
|
|
async def delete_membership(
|
|
username: str,
|
|
groupname: str,
|
|
conn: Annotated[Connection, Depends(get_db_connection)],
|
|
current_user: Annotated[User, Depends(get_current_user)]
|
|
):
|
|
if current_user.role in settings.settings.admin_roles:
|
|
return await db.delete_membership(conn, username, groupname)
|
|
|
|
if await check_group_author(conn, groupname, current_user.username):
|
|
return await db.delete_membership(conn, username, groupname)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Not allowed",
|
|
)
|