sr2/cloud-api
6
0
Fork 1
Code Issues 8 Pull requests 2 Projects Releases Packages Wiki Activity Actions

feat: iam dependencies

Browse source 
IAM endpoints now use dependencies to perform most initial database get requests.

Issue #6
This commit is contained in:
Chris Milne 2026-05-27 11:11:19 +01:00
parent d4f1b73deb
commit c6a2b301dc
4 changed files with 100 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

59
src/iam/router.py
View file

@ -17,6 +17,7 @@ from src.iam.schemas import IAMGetGroupPermissionsResponse, IAMGetGroupUsersResp
IAMPostPermissionResponse, PermissionResponse, IAMDeletePermissionRequest, IAMGetPermissionsSearchRequest, IAMGetPermissionsSearchResponse
from src.schemas import ResourceName
from src.auth.service import claims_dependency
from src.user.exceptions import UserNotFoundException
from src.user.models import User
from src.organisation.models import Organisation as Org
from src.service.models import Service
@ -24,6 +25,7 @@ from src.organisation.dependencies import org_model_dependency
from src.iam.service import service_key_dependency
from src.iam.models import Permission as Perm, GroupPermissions as GPerms, Group, UserGroups
from src.iam.dependencies import group_model_query_dependency, group_model_body_dependency, perm_model_body_dependency
router = APIRouter(
tags=["IAM"],
@ -64,28 +66,21 @@ async def can_act_on_resource(valid_key: service_key_dependency, db: db_dependen
@router.get("/group/permissions", response_model=IAMGetGroupPermissionsResponse)
async def get_group_permissions(db: db_dependency, group_id: Annotated[int, Query(gt=0)]):
# TODO: root_user_dependency & org_id query param
group_model = db.get(Group, group_id)
if group_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
async def get_group_permissions(db: db_dependency, group_model: group_model_query_dependency):
# TODO: root_user_dependency
return {"permissions": group_model.permission_rel}
@router.get("/group/users", response_model=IAMGetGroupUsersResponse)
async def get_group_users(db: db_dependency, group_id: Annotated[int, Query(gt=0)]):
# TODO: root_user_dependency & org_id query param
group_model = db.get(Group, group_id)
if group_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
async def get_group_users(db: db_dependency, group_model: group_model_query_dependency):
# TODO: root_user_dependency
return {"users": group_model.user_rel}
@router.post("/group", response_model=IAMPostGroupResponse)
async def create_group(db: db_dependency, group_request: IAMPostGroupRequest, org_model: org_model_dependency, org_id: Annotated[int, Query(gt=0)]):
# TODO: root_user_dependency
# TODO: get org ID from dependency instead of query (needs updated dep first)
group_model = Group(name=group_request.name, org_id=org_id)
db.add(group_model)
@ -96,15 +91,8 @@ async def create_group(db: db_dependency, group_request: IAMPostGroupRequest, or
@router.put("/group/permission", response_model=IAMPutGroupPermissionResponse)
async def add_group_permission(db: db_dependency, request_model: IAMPutGroupPermissionRequest, org_model: org_model_dependency, org_id: Annotated[int, Query(gt=0)]):
async def add_group_permission(db: db_dependency, group_model: group_model_body_dependency, perm_model: perm_model_body_dependency, request_model: IAMPutGroupPermissionRequest):
# TODO: root_user_dependency
group_model = db.get(Group, request_model.group_id)
if group_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
perm_model = db.get(Perm, request_model.permission_id)
if perm_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Permission not found")
group_model.permission_rel.append(perm_model)
db.flush()
@ -114,14 +102,12 @@ async def add_group_permission(db: db_dependency, request_model: IAMPutGroupPerm
@router.put("/group/user")
async def add_group_user(db: db_dependency, request_model: IAMPutGroupUserRequest, org_model: org_model_dependency, org_id: Annotated[int, Query(gt=0)]):
async def add_group_user(db: db_dependency, group_model: group_model_body_dependency, request_model: IAMPutGroupUserRequest):
# TODO: root_user_dependency
group_model = db.get(Group, request_model.group_id)
if group_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
# TODO: user_model_dependency
user_model = db.get(User, request_model.user_id)
if user_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
raise UserNotFoundException(user_id=request_model.user_id)
group_model.user_rel.append(user_model)
db.flush()
@ -131,15 +117,8 @@ async def add_group_user(db: db_dependency, request_model: IAMPutGroupUserReques
@router.delete("/group/permissions")
async def remove_group_permissions(db: db_dependency, request_model: IAMDeleteGroupPermissionRequest, org_model: org_model_dependency, org_id: Annotated[int, Query(gt=0)]):
async def remove_group_permissions(db: db_dependency, group_model: group_model_body_dependency, perm_model: perm_model_body_dependency, request_model: IAMDeleteGroupPermissionRequest):
# TODO: root_user_dependency
group_model = db.get(Group, request_model.group_id)
if group_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
perm_model = db.get(Perm, request_model.permission_id)
if perm_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Permission not found")
group_model.permission_rel.remove(perm_model)
db.flush()
response = IAMDeleteGroupPermissionResponse(group=GroupResponse(**group_model.__dict__),
@ -149,11 +128,9 @@ async def remove_group_permissions(db: db_dependency, request_model: IAMDeleteGr
@router.delete("/group/user")
async def remove_group_user(db: db_dependency, request_model: IAMDeleteGroupUserRequest, org_model: org_model_dependency, org_id: Annotated[int, Query(gt=0)]):
async def remove_group_user(db: db_dependency, group_model: group_model_body_dependency, request_model: IAMDeleteGroupUserRequest):
# TODO: root_user_dependency
group_model = db.get(Group, request_model.group_id)
if group_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
# TODO: User model dependency
user_model = db.get(User, request_model.user_id)
if user_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
@ -167,7 +144,7 @@ async def remove_group_user(db: db_dependency, request_model: IAMDeleteGroupUser
@router.get("/permissions", response_model=IAMGetPermissionsResponse)
async def get_permissions(db: db_dependency, org_model: org_model_dependency, org_id: Annotated[int, Query(gt=0)]):
async def get_permissions(db: db_dependency):
# TODO: root_user_dependency
permission_models = db.query(Perm).all()
@ -187,12 +164,8 @@ async def create_new_permission(db: db_dependency, request_mode: IAMPostPermissi
@router.delete("/permission", status_code=status.HTTP_204_NO_CONTENT)
async def delete_permission(db: db_dependency, request_model: IAMDeletePermissionRequest):
async def delete_permission(db: db_dependency, perm_model: perm_model_body_dependency, request_model: IAMDeletePermissionRequest):
# TODO: super_admin_dependency
perm_model = db.query(Perm).filter(Perm.service_id==request_model.service_id, Perm.resource==request_model.resource, Perm.action==request_model.action).first()
if perm_model is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Permission not found")
db.delete(perm_model)
db.commit()