docs: user module
In-line and Swagger docs improvements on the User module and endpoints
This commit is contained in:
parent
6871fcd75d
commit
83a24a91f4
3 changed files with 105 additions and 34 deletions
11
src/main.py
11
src/main.py
|
|
@ -26,12 +26,21 @@ if settings.ENVIRONMENT.is_deployed:
|
|||
pass
|
||||
|
||||
|
||||
tags_metadata = [
|
||||
{
|
||||
"name": "User",
|
||||
"description": "User related operations, includes getting information about the current user",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
app = FastAPI(
|
||||
swagger_ui_init_oauth={
|
||||
"clientId": auth_settings.CLIENT_ID,
|
||||
"usePkceWithAuthorizationCodeGrant": True,
|
||||
"scopes": "openid profile email",
|
||||
}
|
||||
},
|
||||
openapi_tags=tags_metadata,
|
||||
)
|
||||
|
||||
# Type inspection disabled for middleware injection.
|
||||
|
|
|
|||
|
|
@ -4,4 +4,16 @@ Module specific exceptions for user module
|
|||
Exceptions:
|
||||
- List: Description
|
||||
- Exceptions: Description
|
||||
"""
|
||||
"""
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
|
||||
|
||||
class UserNotFoundException(HTTPException):
|
||||
def __init__(self, user_id: Optional[int] = None) -> None:
|
||||
detail = "User not found" if user_id is None else f"User with ID '{user_id}' was not found."
|
||||
super().__init__(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=detail,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -2,10 +2,10 @@
|
|||
Router endpoints for user module
|
||||
|
||||
Endpoints:
|
||||
- [get]/me/claims - Retrieves user's OIDC claims
|
||||
- [get]/me/db - Retrieves the user data from the db that corresponds to the current OIDC user
|
||||
- [get]/me/orgs - Retrieves all organisations associated with the current user
|
||||
- [get]/me/orgs/admin - Retrieves only admin organisations for the current user
|
||||
- [get]/self/claims - Retrieves user's OIDC claims
|
||||
- [get]/self/db - Retrieves the user data from the db that corresponds to the current OIDC user
|
||||
- [get]/self/orgs - Retrieves all organisations associated with the current user
|
||||
- [get]/self/orgs/admin - Retrieves only admin organisations for the current user
|
||||
- [get]/{user_id} - Retrieves a specific user by their ID
|
||||
- [get]/{user_id}/orgs - Retrieves all organisations associated with a specific user
|
||||
- [get]/{user_id}/orgs/admin - Retrieves only admin organisations for a specific user
|
||||
|
|
@ -13,12 +13,14 @@ Endpoints:
|
|||
"""
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
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, OIDCUser, OrgResponse
|
||||
from src.user.schemas import UserResponse, OrgResponse, OIDCClaims
|
||||
from src.user.exceptions import UserNotFoundException
|
||||
|
||||
from src.organisation.models import OrgUsers, Organisation
|
||||
|
||||
|
|
@ -27,36 +29,54 @@ from src.database import db_dependency
|
|||
|
||||
router = APIRouter(
|
||||
prefix="/user",
|
||||
tags=["user"],
|
||||
tags=["User"],
|
||||
)
|
||||
|
||||
|
||||
@router.get("/me/claims")
|
||||
@router.get("/self/claims", response_model=OIDCClaims, status_code=status.HTTP_200_OK, responses={
|
||||
status.HTTP_200_OK: {"description": "Successful retrieval from database"},
|
||||
})
|
||||
async def current_user_claims(user: claims_dependency):
|
||||
"""
|
||||
Returns the full OIDC claims associated with the currently logged-in user.
|
||||
"""
|
||||
user["allowed_origins"] = user.get("allowed-origins", [])
|
||||
return user
|
||||
|
||||
|
||||
@router.get("/me/db", response_model=OIDCUser)
|
||||
@router.get("/self/db", 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"},
|
||||
})
|
||||
async def current_user(user: claims_dependency, db: db_dependency):
|
||||
db_id = user.get("db_id", None)
|
||||
if db_id is None:
|
||||
raise HTTPException(status_code=404, detail="User not found in db")
|
||||
"""
|
||||
Returns the database details associated with the currently logged-in user.
|
||||
"""
|
||||
user_id = user.get("db_id", None)
|
||||
if user_id is None:
|
||||
raise UserNotFoundException()
|
||||
|
||||
user_model = (db.query(User).filter(User.id == db_id).first())
|
||||
user_model = (db.query(User).filter(User.id == user_id).first())
|
||||
if user_model is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
raise UserNotFoundException(user_id=user_id)
|
||||
|
||||
return user_model
|
||||
|
||||
|
||||
@router.get("/me/orgs", response_model=list[OrgResponse])
|
||||
@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 HTTPException(status_code=404, detail="User not found")
|
||||
raise UserNotFoundException()
|
||||
user_exists = db.query(exists().where(User.id == user_id)).scalar()
|
||||
if not user_exists:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
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)
|
||||
|
|
@ -67,14 +87,20 @@ async def get_current_organisations(db: db_dependency, user: claims_dependency):
|
|||
return org_user_models
|
||||
|
||||
|
||||
@router.get("/me/orgs/admin", response_model=list[OrgResponse])
|
||||
@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 HTTPException(status_code=404, detail="User not found")
|
||||
raise UserNotFoundException()
|
||||
user_exists = db.query(exists().where(User.id == user_id)).scalar()
|
||||
if not user_exists:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
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)
|
||||
|
|
@ -86,20 +112,32 @@ async def get_current_admin_organisations(db: db_dependency, user: claims_depend
|
|||
return org_user_models
|
||||
|
||||
|
||||
@router.get("/{user_id}", response_model=UserResponse)
|
||||
async def get_user_by_id(user_id: int, db: db_dependency):
|
||||
@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"},
|
||||
})
|
||||
async def get_user_by_id(db: db_dependency, user_id: Annotated[int, Path(gt=0,description="User database ID")]):
|
||||
"""
|
||||
Returns the database details associated with the provided user ID.
|
||||
"""
|
||||
user_model = (db.query(User).filter(User.id == user_id).first())
|
||||
if user_model is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
raise UserNotFoundException(user_id=user_id)
|
||||
|
||||
return user_model
|
||||
|
||||
|
||||
@router.get("/{user_id}/orgs", response_model=list[OrgResponse])
|
||||
async def get_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0)]):
|
||||
@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 HTTPException(status_code=404, detail="User not found")
|
||||
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)
|
||||
|
|
@ -110,11 +148,17 @@ 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])
|
||||
async def get_admin_organisations(db: db_dependency, user_id: Annotated[int, Path(gt=0)]):
|
||||
@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 HTTPException(status_code=404, detail="User not found")
|
||||
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)
|
||||
|
|
@ -126,10 +170,16 @@ async def get_admin_organisations(db: db_dependency, user_id: Annotated[int, Pat
|
|||
return org_user_models
|
||||
|
||||
|
||||
@router.delete("/{user_id}")
|
||||
async def delete_user_by_id(user_id: int, db: db_dependency):
|
||||
@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"},
|
||||
})
|
||||
async def delete_user_by_id(user_id: Annotated[int, Path(gt=0)], db: db_dependency):
|
||||
"""
|
||||
Deletes the user with the provided ID from the database. This will not remove them from OIDC, and they will be automatically readded on next login.
|
||||
"""
|
||||
user_model = (db.query(User).filter(User.id == user_id).first())
|
||||
if user_model is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
raise UserNotFoundException(user_id=user_id)
|
||||
db.delete(user_model)
|
||||
db.commit()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue