feat(admin): Add group admin and test for admin views
This commit is contained in:
@@ -6,11 +6,13 @@ from starlette import status
|
||||
from starlette.responses import RedirectResponse
|
||||
|
||||
from allmende_payment_system.api.dependencies import SessionDep, UserDep
|
||||
from allmende_payment_system.models import User, UserGroup
|
||||
from allmende_payment_system.models import Permission, User, UserGroup
|
||||
from allmende_payment_system.tools import get_jinja_renderer
|
||||
|
||||
admin_router = APIRouter(prefix="/admin")
|
||||
|
||||
# USERS
|
||||
|
||||
|
||||
@admin_router.get("/users")
|
||||
async def user_list(request: Request, session: SessionDep, user: UserDep):
|
||||
@@ -62,3 +64,95 @@ async def user_remove_group(
|
||||
print(user)
|
||||
user.user_groups.remove(group)
|
||||
return RedirectResponse(url="/admin/users", status_code=status.HTTP_303_SEE_OTHER)
|
||||
|
||||
|
||||
# GROUPS
|
||||
|
||||
|
||||
@admin_router.get("/groups")
|
||||
async def group_list(request: Request, session: SessionDep, user: UserDep):
|
||||
if not user.has_permission("user", "edit"):
|
||||
raise HTTPException(status_code=403, detail="Insufficient permissions")
|
||||
|
||||
groups = session.scalars(select(UserGroup)).all()
|
||||
templates = get_jinja_renderer()
|
||||
return templates.TemplateResponse(
|
||||
"groups.html.jinja",
|
||||
context={"request": request, "groups": groups},
|
||||
)
|
||||
|
||||
|
||||
@admin_router.post("/groups/{group_id}/add_permission")
|
||||
async def group_add_permission(
|
||||
request: Request, session: SessionDep, user: UserDep, group_id: int
|
||||
):
|
||||
if not user.has_permission("user", "edit"):
|
||||
raise HTTPException(status_code=403, detail="Insufficient permissions")
|
||||
|
||||
data = await request.form()
|
||||
scope_action = data["permission"].split(":")
|
||||
if len(scope_action) != 2:
|
||||
raise HTTPException(
|
||||
status_code=400, detail="Permission must be in the format 'scope:action'"
|
||||
)
|
||||
|
||||
permission = Permission(scope=scope_action[0], action=scope_action[1])
|
||||
group = session.execute(
|
||||
select(UserGroup).where(UserGroup.id == group_id)
|
||||
).scalar_one()
|
||||
|
||||
session.add(permission)
|
||||
group.permissions.append(permission)
|
||||
|
||||
return RedirectResponse(url="/admin/groups", status_code=status.HTTP_303_SEE_OTHER)
|
||||
|
||||
|
||||
@admin_router.get("/groups/{group_id}/remove_permission/{permission_id}")
|
||||
async def group_remove_permission(
|
||||
request: Request,
|
||||
session: SessionDep,
|
||||
user: UserDep,
|
||||
group_id: int,
|
||||
permission_id: int,
|
||||
):
|
||||
if not user.has_permission("user", "edit"):
|
||||
raise HTTPException(status_code=403, detail="Insufficient permissions")
|
||||
|
||||
permission = session.execute(
|
||||
select(Permission).where(Permission.id == permission_id)
|
||||
).scalar_one()
|
||||
group = session.execute(
|
||||
select(UserGroup).where(UserGroup.id == group_id)
|
||||
).scalar_one()
|
||||
group.permissions.remove(permission)
|
||||
session.delete(permission)
|
||||
return RedirectResponse(url="/admin/groups", status_code=status.HTTP_303_SEE_OTHER)
|
||||
|
||||
|
||||
@admin_router.post("/groups/create")
|
||||
async def create_group(request: Request, session: SessionDep, user: UserDep):
|
||||
if not user.has_permission("user", "edit"):
|
||||
raise HTTPException(status_code=403, detail="Insufficient permissions")
|
||||
|
||||
data = await request.form()
|
||||
group = UserGroup(name=data["name"], description=data["description"])
|
||||
session.add(group)
|
||||
|
||||
return RedirectResponse(url="/admin/groups", status_code=status.HTTP_303_SEE_OTHER)
|
||||
|
||||
|
||||
@admin_router.get("/groups/{group_id}/delete")
|
||||
async def delete_group(
|
||||
request: Request,
|
||||
session: SessionDep,
|
||||
user: UserDep,
|
||||
group_id: int,
|
||||
):
|
||||
if not user.has_permission("user", "edit"):
|
||||
raise HTTPException(status_code=403, detail="Insufficient permissions")
|
||||
|
||||
group = session.execute(
|
||||
select(UserGroup).where(UserGroup.id == group_id)
|
||||
).scalar_one()
|
||||
session.delete(group)
|
||||
return RedirectResponse(url="/admin/groups", status_code=status.HTTP_303_SEE_OTHER)
|
||||
|
||||
Reference in New Issue
Block a user