145 lines
4.6 KiB
Python
145 lines
4.6 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth.jwt_auth import CurrentUser, require_admin
|
|
from app.database import get_db
|
|
from app.models.user import USER_ROLE_ADMIN, USER_ROLE_MONITOR, USER_ROLES, User
|
|
from app.services.user_auth import create_user, hash_password
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
|
|
class UserSummary(BaseModel):
|
|
id: int
|
|
username: str
|
|
role: str
|
|
is_active: bool
|
|
created_at: datetime
|
|
|
|
|
|
class UserListResponse(BaseModel):
|
|
items: list[UserSummary]
|
|
|
|
|
|
class CreateUserRequest(BaseModel):
|
|
username: str = Field(min_length=2, max_length=64)
|
|
password: str = Field(min_length=8, max_length=128)
|
|
role: str = Field(default=USER_ROLE_MONITOR)
|
|
|
|
|
|
class UpdateUserRequest(BaseModel):
|
|
username: str | None = Field(default=None, min_length=2, max_length=64)
|
|
role: str | None = None
|
|
password: str | None = Field(default=None, min_length=8, max_length=128)
|
|
is_active: bool | None = None
|
|
|
|
|
|
@router.get("", response_model=UserListResponse)
|
|
def list_users(
|
|
_admin: CurrentUser = Depends(require_admin),
|
|
db: Session = Depends(get_db),
|
|
) -> UserListResponse:
|
|
rows = db.scalars(select(User).order_by(User.username)).all()
|
|
return UserListResponse(
|
|
items=[
|
|
UserSummary(
|
|
id=u.id,
|
|
username=u.username,
|
|
role=u.role,
|
|
is_active=u.is_active,
|
|
created_at=u.created_at,
|
|
)
|
|
for u in rows
|
|
]
|
|
)
|
|
|
|
|
|
@router.post("", response_model=UserSummary, status_code=201)
|
|
def create_user_endpoint(
|
|
body: CreateUserRequest,
|
|
admin: CurrentUser = Depends(require_admin),
|
|
db: Session = Depends(get_db),
|
|
) -> UserSummary:
|
|
if body.role not in USER_ROLES:
|
|
raise HTTPException(status_code=422, detail=f"role must be one of: {', '.join(sorted(USER_ROLES))}")
|
|
try:
|
|
user = create_user(db, username=body.username, password=body.password, role=body.role)
|
|
db.commit()
|
|
db.refresh(user)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
return UserSummary(
|
|
id=user.id,
|
|
username=user.username,
|
|
role=user.role,
|
|
is_active=user.is_active,
|
|
created_at=user.created_at,
|
|
)
|
|
|
|
|
|
@router.patch("/{user_id}", response_model=UserSummary)
|
|
def update_user(
|
|
user_id: int,
|
|
body: UpdateUserRequest,
|
|
admin: CurrentUser = Depends(require_admin),
|
|
db: Session = Depends(get_db),
|
|
) -> UserSummary:
|
|
user = db.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
if body.username is not None:
|
|
normalized = body.username.strip()
|
|
if not normalized:
|
|
raise HTTPException(status_code=422, detail="username is required")
|
|
existing = db.scalar(
|
|
select(User).where(
|
|
func.lower(User.username) == normalized.lower(),
|
|
User.id != user.id,
|
|
)
|
|
)
|
|
if existing is not None:
|
|
raise HTTPException(status_code=400, detail="username already exists")
|
|
user.username = normalized
|
|
|
|
if body.role is not None:
|
|
if body.role not in USER_ROLES:
|
|
raise HTTPException(status_code=422, detail=f"role must be one of: {', '.join(sorted(USER_ROLES))}")
|
|
if user.role == USER_ROLE_ADMIN and body.role != USER_ROLE_ADMIN:
|
|
_ensure_not_last_admin(db, user.id)
|
|
user.role = body.role
|
|
|
|
if body.password is not None:
|
|
user.password_hash = hash_password(body.password)
|
|
|
|
if body.is_active is not None:
|
|
if not body.is_active and user.username.lower() == admin.username.lower():
|
|
raise HTTPException(status_code=400, detail="Cannot deactivate your own account")
|
|
if not body.is_active and user.role == USER_ROLE_ADMIN:
|
|
_ensure_not_last_admin(db, user.id)
|
|
user.is_active = body.is_active
|
|
|
|
db.commit()
|
|
db.refresh(user)
|
|
return UserSummary(
|
|
id=user.id,
|
|
username=user.username,
|
|
role=user.role,
|
|
is_active=user.is_active,
|
|
created_at=user.created_at,
|
|
)
|
|
|
|
|
|
def _ensure_not_last_admin(db: Session, user_id: int) -> None:
|
|
active_admins = db.scalar(
|
|
select(func.count())
|
|
.select_from(User)
|
|
.where(User.role == USER_ROLE_ADMIN, User.is_active.is_(True), User.id != user_id)
|
|
)
|
|
if not active_admins:
|
|
raise HTTPException(status_code=400, detail="Cannot remove or deactivate the last admin user")
|