125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""User management API routes."""
|
|
|
|
from __future__ import annotations
|
|
from typing import NoReturn
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Response, status
|
|
|
|
from app.core.auth import AuthContext, require_admin
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.models import User
|
|
from app.repositories.factory import get_portfolio_repository_from_settings
|
|
from app.schemas.user import CreateUserRequest, UpdateUserRequest, UserResponse
|
|
from app.services.user_service import UserService
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
|
|
def _get_service(settings: Settings) -> tuple[UserService, object]:
|
|
repository = get_portfolio_repository_from_settings(settings)
|
|
return UserService(repository), repository
|
|
|
|
|
|
def _close_repository(repository: object) -> None:
|
|
if callable(getattr(type(repository), "close", None)):
|
|
repository.close()
|
|
|
|
|
|
def _to_user_response(user: User) -> UserResponse:
|
|
return UserResponse(
|
|
id=user.id,
|
|
username=user.username,
|
|
email=user.email,
|
|
permission=user.permission,
|
|
is_active=user.is_active,
|
|
created_at=user.created_at,
|
|
updated_at=user.updated_at,
|
|
last_login_at=user.last_login_at,
|
|
)
|
|
|
|
|
|
def _raise_user_error(exc: ValueError) -> NoReturn:
|
|
detail = str(exc)
|
|
if "찾을 수 없습니다" in detail:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=detail) from exc
|
|
if "이미 존재" in detail:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=detail) from exc
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail) from exc
|
|
|
|
|
|
@router.get("", response_model=list[UserResponse])
|
|
def list_users(
|
|
_auth: AuthContext = Depends(require_admin),
|
|
settings: Settings = Depends(get_settings),
|
|
) -> list[UserResponse]:
|
|
service, repository = _get_service(settings)
|
|
try:
|
|
return [_to_user_response(user) for user in service.list()]
|
|
finally:
|
|
_close_repository(repository)
|
|
|
|
|
|
@router.get("/{user_id}", response_model=UserResponse)
|
|
def get_user(
|
|
user_id: str,
|
|
_auth: AuthContext = Depends(require_admin),
|
|
settings: Settings = Depends(get_settings),
|
|
) -> UserResponse:
|
|
try:
|
|
service, repository = _get_service(settings)
|
|
try:
|
|
return _to_user_response(service.get(user_id))
|
|
finally:
|
|
_close_repository(repository)
|
|
except ValueError as exc:
|
|
_raise_user_error(exc)
|
|
|
|
|
|
@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_user(
|
|
payload: CreateUserRequest,
|
|
_auth: AuthContext = Depends(require_admin),
|
|
settings: Settings = Depends(get_settings),
|
|
) -> UserResponse:
|
|
try:
|
|
service, repository = _get_service(settings)
|
|
try:
|
|
return _to_user_response(service.create(payload))
|
|
finally:
|
|
_close_repository(repository)
|
|
except ValueError as exc:
|
|
_raise_user_error(exc)
|
|
|
|
|
|
@router.patch("/{user_id}", response_model=UserResponse)
|
|
def update_user(
|
|
user_id: str,
|
|
payload: UpdateUserRequest,
|
|
_auth: AuthContext = Depends(require_admin),
|
|
settings: Settings = Depends(get_settings),
|
|
) -> UserResponse:
|
|
try:
|
|
service, repository = _get_service(settings)
|
|
try:
|
|
return _to_user_response(service.update(user_id, payload))
|
|
finally:
|
|
_close_repository(repository)
|
|
except ValueError as exc:
|
|
_raise_user_error(exc)
|
|
|
|
|
|
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_user(
|
|
user_id: str,
|
|
_auth: AuthContext = Depends(require_admin),
|
|
settings: Settings = Depends(get_settings),
|
|
) -> Response:
|
|
try:
|
|
service, repository = _get_service(settings)
|
|
try:
|
|
service.delete(user_id)
|
|
finally:
|
|
_close_repository(repository)
|
|
except ValueError as exc:
|
|
_raise_user_error(exc)
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT) |