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

feat: iam rbac system

Browse source 
Endpoints and db architecture to support a role based IAM system.
This commit is contained in:
Chris Milne 2026-05-25 09:05:17 +01:00
parent 7b3ee9d5fa
commit 23f2ce98d7
31 changed files with 634 additions and 317 deletions
Download patch file Download diff file Expand all files Collapse all files

65
src/user/router.py
View file

@ -23,7 +23,7 @@ from src.user.schemas import UserResponse, OrgResponse, OIDCClaims
from src.user.exceptions import UserNotFoundException
from src.organisation.models import OrgUsers, Organisation
from src.iam.models import Group, UserGroups
from src.auth.service import claims_dependency
from src.database import db_dependency
@ -78,7 +78,7 @@ async def get_current_organisations(db: db_dependency, user: claims_dependency):
if not user_exists:
raise UserNotFoundException(user_id=user_id)
org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name)
org_user_models = (db.query(OrgUsers.org_id, Organisation.name)
.join(OrgUsers, Organisation.id == OrgUsers.org_id)
.filter(OrgUsers.user_id == user_id)
.all()
@ -87,31 +87,6 @@ async def get_current_organisations(db: db_dependency, user: claims_dependency):
return org_user_models
@router.get("/self/orgs/admin", response_model=list[OrgResponse], status_code=status.HTTP_200_OK, responses={
status.HTTP_404_NOT_FOUND: {"description": "User not found"},
status.HTTP_200_OK: {"description": "Successful retrieval from database"},
})
async def get_current_admin_organisations(db: db_dependency, user: claims_dependency):
"""
Returns the organisations for which the currently logged-in user is an admin.
"""
user_id = user.get("db_id", None)
if user_id is None:
raise UserNotFoundException()
user_exists = db.query(exists().where(User.id == user_id)).scalar()
if not user_exists:
raise UserNotFoundException(user_id=user_id)
org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name)
.join(OrgUsers, Organisation.id == OrgUsers.org_id)
.filter(OrgUsers.user_id == user_id)
.filter(OrgUsers.is_admin == True)
.all()
)
return org_user_models
@router.get("/{user_id}", response_model=UserResponse, status_code=status.HTTP_200_OK, responses={
status.HTTP_404_NOT_FOUND: {"description": "User not found"},
status.HTTP_200_OK: {"description": "Successful retrieval from database"},
@ -139,7 +114,7 @@ async def get_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0
if not user_exists:
raise UserNotFoundException(user_id=user_id)
org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name)
org_user_models = (db.query(OrgUsers.org_id, Organisation.name)
.join(OrgUsers, Organisation.id == OrgUsers.org_id)
.filter(OrgUsers.user_id == user_id)
.all()
@ -148,28 +123,6 @@ async def get_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0
return org_user_models
@router.get("/{user_id}/orgs/admin", response_model=list[OrgResponse], status_code=status.HTTP_200_OK, responses={
status.HTTP_404_NOT_FOUND: {"description": "User not found"},
status.HTTP_200_OK: {"description": "Successful retrieval from database"},
})
async def get_admin_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]):
"""
Returns the organisations for which the user with the provided user ID is an admin.
"""
user_exists = db.query(exists().where(User.id == user_id)).scalar()
if not user_exists:
raise UserNotFoundException(user_id=user_id)
org_user_models = (db.query(OrgUsers.org_id, OrgUsers.is_admin, Organisation.name)
.join(OrgUsers, Organisation.id == OrgUsers.org_id)
.filter(OrgUsers.user_id == user_id)
.filter(OrgUsers.is_admin == True)
.all()
)
return org_user_models
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT, responses={
status.HTTP_204_NO_CONTENT: {"description": "User deleted"},
status.HTTP_404_NOT_FOUND: {"description": "User not found"},
@ -183,3 +136,15 @@ async def delete_user_by_id(user_id: Annotated[int, Path(gt=0)], db: db_dependen
raise UserNotFoundException(user_id=user_id)
db.delete(user_model)
db.commit()
@router.get("/{user_id}/groups")
async def get_user_groups(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]):
user_model = (db.query(User).filter(User.id == user_id).first())
if user_model is None:
raise UserNotFoundException(user_id=user_id)
user_groups = db.query(Group).join(UserGroups).filter(UserGroups.user_id==user_id).all()
# TODO: Response model
return user_groups

1
src/user/schemas.py
View file

@ -49,4 +49,3 @@ class UserResponse(CustomBaseModel):
class OrgResponse(CustomBaseModel):
org_id: int
name: str
is_admin: bool