61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
import logging
|
|
import re
|
|
|
|
from fastapi import APIRouter, HTTPException, status
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app.api.dependencies import CurrentOwnerUser, SessionDep
|
|
from app.crud import user_crud
|
|
from app.database.models.user_model import UserRole
|
|
from app.schemas.user_schemas import UserPublic, UserRegister, UserUpdate
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
INTERNAL_SERVER_ERROR = HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error")
|
|
|
|
|
|
router = APIRouter(
|
|
prefix="/user",
|
|
tags=["User", "Dashboard"]
|
|
)
|
|
|
|
|
|
@router.get("", summary="Get information about currently logged in user")
|
|
async def get_user(current_user: CurrentOwnerUser) -> UserPublic:
|
|
return current_user
|
|
|
|
|
|
@router.post("", summary="Register new user", status_code=status.HTTP_201_CREATED, response_model=bool)
|
|
async def register(session: SessionDep, user_data: UserRegister):
|
|
try:
|
|
user_crud.create_user(session, user_data, None, UserRole.OWNER)
|
|
return True
|
|
except IntegrityError as e:
|
|
field_mapping = {"uuid": "email"} # If a UUID is duplicate, it means email is in use
|
|
|
|
constraint = e.orig.diag.constraint_name
|
|
column_name = re.sub(r"^user_(\w+)_key$", r"\1", constraint) if constraint else None
|
|
|
|
if column_name == "uuid":
|
|
column_name = "email"
|
|
|
|
detail = f"{field_mapping.get(column_name, column_name or 'Entry').capitalize()} already in use"
|
|
logger.warning("%s already exists in the database", column_name.capitalize)
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=detail)
|
|
except Exception as e:
|
|
if isinstance(e, HTTPException):
|
|
raise
|
|
logger.error(e)
|
|
raise INTERNAL_SERVER_ERROR
|
|
|
|
|
|
@router.delete("", summary="Delete user")
|
|
async def delete_user() -> bool:
|
|
raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="delete_user() not implemented")
|
|
|
|
|
|
@router.put("", summary="Update user details")
|
|
async def update_user(session: SessionDep, data: UserUpdate, current_user: CurrentOwnerUser) -> bool:
|
|
user_crud.update_user(session, data, current_user)
|
|
return True
|