feat: condensed user get endpoints
The process also added improved ORM relationships for multiple models.
This commit is contained in:
parent
4ff184fe86
commit
a80767d870
5 changed files with 39 additions and 65 deletions
|
|
@ -4,10 +4,13 @@ Database models for user module
|
|||
Models:
|
||||
- User - id[pk], email, first_name, last_name, oidc_id
|
||||
"""
|
||||
from collections import defaultdict
|
||||
|
||||
from sqlalchemy import Column, Integer, String
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from src.database import Base
|
||||
from src.iam.models import Group
|
||||
|
||||
|
||||
class User(Base):
|
||||
|
|
@ -19,8 +22,25 @@ class User(Base):
|
|||
last_name = Column(String)
|
||||
oidc_id = Column(String, index=True, unique=True)
|
||||
|
||||
organisations = relationship(
|
||||
organisation_rel = relationship(
|
||||
"Organisation",
|
||||
secondary="orgusers",
|
||||
back_populates="users"
|
||||
back_populates="user_rel"
|
||||
)
|
||||
|
||||
@property
|
||||
def organisations(self):
|
||||
return [org.name for org in self.organisation_rel]
|
||||
|
||||
group_rel = relationship(
|
||||
"Group",
|
||||
secondary="user_groups",
|
||||
back_populates="user_rel"
|
||||
)
|
||||
|
||||
@property
|
||||
def groups(self):
|
||||
result = defaultdict(list)
|
||||
for group in self.group_rel:
|
||||
result[group.org_rel.name].append(group.name)
|
||||
return dict(result)
|
||||
|
|
|
|||
|
|
@ -15,15 +15,12 @@ from typing import Annotated
|
|||
|
||||
from fastapi import APIRouter
|
||||
from fastapi.params import Path
|
||||
from sqlalchemy.sql import exists
|
||||
from starlette import status
|
||||
|
||||
from src.user.models import User
|
||||
from src.user.schemas import UserResponse, OrgResponse, OIDCClaims
|
||||
from src.user.schemas import UserResponse, 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
|
||||
|
||||
|
|
@ -63,30 +60,6 @@ async def current_user(user: claims_dependency, db: db_dependency):
|
|||
return user_model
|
||||
|
||||
|
||||
@router.get("/self/orgs", 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_organisations(db: db_dependency, user: claims_dependency):
|
||||
"""
|
||||
Returns all organisations associated with the currently logged-in user.
|
||||
"""
|
||||
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, Organisation.name)
|
||||
.join(OrgUsers, Organisation.id == OrgUsers.org_id)
|
||||
.filter(OrgUsers.user_id == user_id)
|
||||
.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"},
|
||||
|
|
@ -102,27 +75,6 @@ async def get_user_by_id(db: db_dependency, user_id: Annotated[int, Path(gt=0,de
|
|||
return user_model
|
||||
|
||||
|
||||
@router.get("/{user_id}/orgs", 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_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]):
|
||||
"""
|
||||
Returns all organisations associated with the provided user ID.
|
||||
"""
|
||||
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, Organisation.name)
|
||||
.join(OrgUsers, Organisation.id == OrgUsers.org_id)
|
||||
.filter(OrgUsers.user_id == user_id)
|
||||
.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"},
|
||||
|
|
@ -136,15 +88,3 @@ 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
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ Models:
|
|||
- List: Description
|
||||
- Models: Description
|
||||
"""
|
||||
from typing import Optional
|
||||
from src.schemas import CustomBaseModel
|
||||
|
||||
|
||||
|
|
@ -44,6 +45,8 @@ class UserResponse(CustomBaseModel):
|
|||
first_name: str
|
||||
last_name: str
|
||||
email: str
|
||||
organisations: list[Optional[str]]
|
||||
groups: dict[str, list[str]]
|
||||
|
||||
|
||||
class OrgResponse(CustomBaseModel):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue