lauadminandClaude Opus 5.5
b8f2fe6b6e
- All frontend pages, labels, notices and errors; html lang=uk, uk-UA money format - Brand "Assistant System" in the top bar and page title - Backend error details returned to the UI (auth, orders, receipts, cash registers, Checkbox/CRM/NP errors) and CLI output - Tests updated for the new messages Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
"""Управление пользователями. Доступно только администратору."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from pydantic import BaseModel, ConfigDict, EmailStr, Field
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app.api.deps import CurrentUser, SessionDep, require_admin
|
|
from app.core.security import hash_password
|
|
from app.db.models.audit import AuditAction
|
|
from app.db.models.user import User, UserRole
|
|
from app.schemas.auth import UserOut
|
|
from app.services import audit, auth
|
|
|
|
router = APIRouter(
|
|
prefix="/users", tags=["users"], dependencies=[Depends(require_admin)]
|
|
)
|
|
|
|
|
|
class UserCreate(BaseModel):
|
|
email: EmailStr
|
|
full_name: str = Field(min_length=1, max_length=255)
|
|
password: str = Field(min_length=10, max_length=128)
|
|
role: UserRole = UserRole.CASHIER
|
|
|
|
|
|
class UserUpdate(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
full_name: str | None = Field(default=None, min_length=1, max_length=255)
|
|
role: UserRole | None = None
|
|
is_active: bool | None = None
|
|
password: str | None = Field(default=None, min_length=10, max_length=128)
|
|
|
|
|
|
@router.get("", response_model=list[UserOut])
|
|
async def list_users(session: SessionDep) -> list[UserOut]:
|
|
users = await session.scalars(select(User).order_by(User.created_at.desc()))
|
|
return [UserOut.model_validate(user) for user in users]
|
|
|
|
|
|
@router.post("", response_model=UserOut, status_code=status.HTTP_201_CREATED)
|
|
async def create_user(
|
|
payload: UserCreate, request: Request, session: SessionDep, actor: CurrentUser
|
|
) -> UserOut:
|
|
user = User(
|
|
email=payload.email.strip().lower(),
|
|
full_name=payload.full_name,
|
|
password_hash=hash_password(payload.password),
|
|
role=payload.role,
|
|
)
|
|
session.add(user)
|
|
|
|
await audit.record(
|
|
session,
|
|
action=AuditAction.USER_CREATED,
|
|
user=actor,
|
|
entity_type="user",
|
|
entity_id=user.id,
|
|
payload={"email": user.email, "role": user.role.value},
|
|
request=request,
|
|
)
|
|
|
|
try:
|
|
await session.commit()
|
|
except IntegrityError as exc:
|
|
await session.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Користувач із таким email уже існує",
|
|
) from exc
|
|
|
|
return UserOut.model_validate(user)
|
|
|
|
|
|
@router.patch("/{user_id}", response_model=UserOut)
|
|
async def update_user(
|
|
user_id: uuid.UUID,
|
|
payload: UserUpdate,
|
|
request: Request,
|
|
session: SessionDep,
|
|
actor: CurrentUser,
|
|
) -> UserOut:
|
|
user = await session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Користувача не знайдено")
|
|
|
|
changes = payload.model_dump(exclude_unset=True)
|
|
|
|
# Администратор, снявший с себя роль или отключивший себя, потеряет доступ
|
|
# к админке — и вернуть его будет некому.
|
|
if user.id == actor.id:
|
|
if changes.get("is_active") is False:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Не можна вимкнути власний обліковий запис",
|
|
)
|
|
if "role" in changes and changes["role"] != UserRole.ADMIN:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Не можна зняти із себе роль адміністратора",
|
|
)
|
|
|
|
if (new_password := changes.pop("password", None)) is not None:
|
|
user.password_hash = hash_password(new_password)
|
|
await auth.revoke_all_for_user(session, user.id)
|
|
|
|
for field, value in changes.items():
|
|
setattr(user, field, value)
|
|
|
|
# Отключение учётки обязано немедленно завершить её активные сессии.
|
|
if changes.get("is_active") is False:
|
|
await auth.revoke_all_for_user(session, user.id)
|
|
|
|
await audit.record(
|
|
session,
|
|
action=AuditAction.USER_UPDATED,
|
|
user=actor,
|
|
entity_type="user",
|
|
entity_id=user.id,
|
|
payload={
|
|
"fields": sorted(changes),
|
|
"password_changed": new_password is not None,
|
|
},
|
|
request=request,
|
|
)
|
|
await session.commit()
|
|
|
|
return UserOut.model_validate(user)
|