218 lines
7.9 KiB
Python
218 lines
7.9 KiB
Python
"""Users router - CRUD operations (admin only)."""
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.backend.database import get_db
|
|
from src.backend.api.middleware.api_key import require_admin_user
|
|
from src.backend.models.orm.user import User
|
|
from src.backend.models.api.user import PinSet, UserCreate, UserPasswordChange, UserResponse, UserUpdate
|
|
from src.backend.services.auth_service import (
|
|
create_user,
|
|
hash_password,
|
|
regenerate_api_key,
|
|
revoke_all_keys,
|
|
set_user_pin,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/users", tags=["users"])
|
|
|
|
# Combinable user roles. Supervisor (capoturno) authorizes out-of-tolerance
|
|
# overrides during measurement without needing full admin privileges.
|
|
VALID_ROLES = {"Maker", "MeasurementTec", "Metrologist", "Supervisor"}
|
|
|
|
|
|
@router.get("", response_model=list[UserResponse])
|
|
async def list_users(
|
|
admin: User = Depends(require_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""List all users (admin only)."""
|
|
result = await db.execute(select(User).order_by(User.username))
|
|
users = result.scalars().all()
|
|
return [UserResponse.model_validate(u) for u in users]
|
|
|
|
|
|
@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
|
|
async def create_new_user(
|
|
data: UserCreate,
|
|
admin: User = Depends(require_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Create a new user (admin only)."""
|
|
# Check username uniqueness
|
|
existing = await db.execute(
|
|
select(User).where(User.username == data.username)
|
|
)
|
|
if existing.scalar_one_or_none():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"Username '{data.username}' already exists",
|
|
)
|
|
# Validate roles
|
|
for role in data.roles:
|
|
if role not in VALID_ROLES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Invalid role '{role}'. Valid roles: {VALID_ROLES}",
|
|
)
|
|
user = await create_user(
|
|
db,
|
|
username=data.username,
|
|
password=data.password,
|
|
display_name=data.display_name,
|
|
email=data.email,
|
|
roles=data.roles,
|
|
is_admin=data.is_admin,
|
|
language_pref=data.language_pref,
|
|
theme_pref=data.theme_pref,
|
|
)
|
|
return UserResponse.model_validate(user)
|
|
|
|
|
|
@router.put("/{user_id}", response_model=UserResponse)
|
|
async def update_user(
|
|
user_id: int,
|
|
data: UserUpdate,
|
|
admin: User = Depends(require_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Update user (admin only)."""
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
|
|
# roles and is_admin are NOT NULL columns: an explicit null passed the
|
|
# Optional[...] schema check but blew up at flush time as an unhandled
|
|
# 500 (#36 point 4).
|
|
if "roles" in update_data and update_data["roles"] is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="roles cannot be null",
|
|
)
|
|
if "is_admin" in update_data and update_data["is_admin"] is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="is_admin cannot be null",
|
|
)
|
|
|
|
# Validate roles if provided
|
|
if "roles" in update_data:
|
|
for role in update_data["roles"]:
|
|
if role not in VALID_ROLES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Invalid role '{role}'. Valid roles: {VALID_ROLES}",
|
|
)
|
|
|
|
# Losing the last admin locks everyone out of user management; losing
|
|
# your own admin flag while acting as admin is the same mistake aimed
|
|
# at yourself. Neither is reversible without direct DB access.
|
|
if update_data.get("is_admin") is False and user.is_admin:
|
|
if user.id == admin.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Cannot remove your own admin privilege",
|
|
)
|
|
other_admins = await db.execute(
|
|
select(func.count()).select_from(User).where(
|
|
User.is_admin == True, User.id != user.id # noqa: E712
|
|
)
|
|
)
|
|
if other_admins.scalar_one() == 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Cannot remove admin privilege from the last admin",
|
|
)
|
|
|
|
for field, value in update_data.items():
|
|
setattr(user, field, value)
|
|
await db.flush()
|
|
await db.refresh(user)
|
|
return UserResponse.model_validate(user)
|
|
|
|
|
|
@router.put("/{user_id}/password", status_code=status.HTTP_200_OK)
|
|
async def change_user_password(
|
|
user_id: int,
|
|
data: UserPasswordChange,
|
|
admin: User = Depends(require_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Change user password (admin only). Invalidates the user's API key,
|
|
|
|
the same as setup.py's change-user-password: a password change should
|
|
force everyone logged in as that user to log back in.
|
|
"""
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
user.password_hash = hash_password(data.password)
|
|
await revoke_all_keys(db, user.id)
|
|
await db.flush()
|
|
return {"message": f"Password changed for user {user.username}"}
|
|
|
|
|
|
@router.put("/{user_id}/pin", status_code=status.HTTP_200_OK)
|
|
async def change_user_pin(
|
|
user_id: int,
|
|
data: PinSet,
|
|
admin: User = Depends(require_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Imposta o toglie (pin null) il PIN di un utente (admin, #59/#51).
|
|
|
|
422 se l'utente e' admin. Il PIN non si restituisce mai.
|
|
"""
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
if user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Admin users cannot have a PIN",
|
|
)
|
|
await set_user_pin(db, user, data.pin)
|
|
return {"message": f"PIN {'set' if data.pin else 'removed'} for user {user.username}"}
|
|
|
|
|
|
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def deactivate_user(
|
|
user_id: int,
|
|
admin: User = Depends(require_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Soft-delete user (set active=False)."""
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
if user.id == admin.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Cannot deactivate yourself",
|
|
)
|
|
user.active = False
|
|
await revoke_all_keys(db, user.id)
|
|
await db.flush()
|
|
|
|
|
|
@router.post("/{user_id}/regenerate-key")
|
|
async def regenerate_user_key(
|
|
user_id: int,
|
|
admin: User = Depends(require_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Regenerate API key for a user (admin only)."""
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
new_key = await regenerate_api_key(db, user)
|
|
return {"api_key": new_key, "message": f"API key regenerated for user {user.username}"}
|