Agent Zero
148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
"""User API endpoints: me, list, create, update, soft-delete."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.db import get_db
|
|
from app.core.deps import get_current_admin_user, get_current_user
|
|
from app.models.user import User, UserRole
|
|
from app.schemas.user import (
|
|
UserCreateRequest,
|
|
UserListResponse,
|
|
UserOut,
|
|
UserUpdate,
|
|
)
|
|
from app.services import user_service
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
|
|
@router.get(
|
|
"/me",
|
|
response_model=UserOut,
|
|
summary="Get the currently authenticated user",
|
|
)
|
|
async def get_me(
|
|
current_user: User = Depends(get_current_user),
|
|
) -> UserOut:
|
|
"""Returns the user from the JWT. Used by the frontend for auth checks and profile display."""
|
|
return UserOut.model_validate(current_user)
|
|
|
|
|
|
@router.patch(
|
|
"/me",
|
|
response_model=UserOut,
|
|
summary="Update own profile (name, avatar, notification prefs)",
|
|
)
|
|
async def update_me(
|
|
payload: UserUpdate,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> UserOut:
|
|
"""A user can update their own profile, but not their role."""
|
|
updated = await user_service.update_user_profile(db, current_user, payload, is_admin=False)
|
|
return UserOut.model_validate(updated)
|
|
|
|
|
|
@router.get(
|
|
"/",
|
|
response_model=UserListResponse,
|
|
summary="List all users in the current org (admin only)",
|
|
)
|
|
async def list_users(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=100),
|
|
current_user: User = Depends(get_current_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> UserListResponse:
|
|
"""Paginated list of active users. Restricted to admin role."""
|
|
skip = (page - 1) * page_size
|
|
users = await user_service.list_users(
|
|
db, org_id=current_user.org_id, skip=skip, limit=page_size
|
|
)
|
|
total = await user_service.count_users_in_org(db, current_user.org_id)
|
|
return UserListResponse(
|
|
items=[UserOut.model_validate(u) for u in users],
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/",
|
|
response_model=UserOut,
|
|
status_code=status.HTTP_201_CREATED,
|
|
summary="Create a new user in the current org (admin only)",
|
|
)
|
|
async def create_user(
|
|
payload: UserCreateRequest,
|
|
current_user: User = Depends(get_current_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> UserOut:
|
|
"""Admin creates a new user in the same org."""
|
|
try:
|
|
new_user = await user_service.create_user_as_admin(db, payload, org_id=current_user.org_id)
|
|
except user_service.EmailAlreadyTaken as e:
|
|
raise HTTPException(status_code=409, detail=str(e)) from e
|
|
return UserOut.model_validate(new_user)
|
|
|
|
|
|
@router.patch(
|
|
"/{user_id}",
|
|
response_model=UserOut,
|
|
summary="Update a user (admin or self for non-role fields)",
|
|
)
|
|
async def update_user(
|
|
user_id: int,
|
|
payload: UserUpdate,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> UserOut:
|
|
"""Admin can update any user (including role); non-admin can update only their own profile fields."""
|
|
is_admin = (
|
|
current_user.role == UserRole.admin
|
|
if hasattr(current_user.role, "__eq__") and not isinstance(current_user.role, str)
|
|
else str(current_user.role) == UserRole.admin.value
|
|
)
|
|
if not is_admin and current_user.id != user_id:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="You can only update your own profile",
|
|
)
|
|
|
|
target = await user_service.get_user_by_id(db, user_id)
|
|
if target is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
if target.org_id != current_user.org_id:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
updated = await user_service.update_user_profile(db, target, payload, is_admin=is_admin)
|
|
return UserOut.model_validate(updated)
|
|
|
|
|
|
@router.delete(
|
|
"/{user_id}",
|
|
status_code=status.HTTP_204_NO_CONTENT,
|
|
response_class=Response,
|
|
summary="Soft-delete a user (admin only)",
|
|
)
|
|
async def delete_user(
|
|
user_id: int,
|
|
current_user: User = Depends(get_current_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> Response:
|
|
"""Sets deleted_at timestamp. The record stays in the DB for audit purposes."""
|
|
if current_user.id == user_id:
|
|
raise HTTPException(status_code=400, detail="You cannot delete your own account")
|
|
|
|
target = await user_service.get_user_by_id(db, user_id)
|
|
if target is None or target.org_id != current_user.org_id:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
await user_service.soft_delete_user(db, target)
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|