Fix: 업로드 및 API 연결 문제 해결
- FastAPI 라우터에서 슬래시 문제로 인한 307 리다이렉트 수정 - Nginx 프록시 설정에서 경로 중복 문제 해결 - 계정 관리 시스템 구현 (로그인, 사용자 관리, 권한 설정) - 노트북 연결 기능 수정 (notebook_id 필드 추가) - 메모 트리 UI 개선 (수평 레이아웃, 드래그 기능 제거) - 헤더 UI 개선 및 고정 위치 설정 - 백업/복원 스크립트 추가 - PDF 미리보기 토큰 인증 지원
This commit is contained in:
Hyungi Ahn
@@ -1,92 +1,276 @@
|
||||
"""
|
||||
사용자 관리 API 라우터
|
||||
사용자 관리 API
|
||||
"""
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select, update, delete
|
||||
from typing import List
|
||||
from sqlalchemy.orm import selectinload
|
||||
from pydantic import BaseModel, EmailStr
|
||||
from typing import List, Optional
|
||||
from datetime import datetime
|
||||
|
||||
from ...core.database import get_db
|
||||
from ...core.security import get_password_hash, verify_password
|
||||
from ...models.user import User
|
||||
from ...schemas.auth import UserInfo
|
||||
from ..dependencies import get_current_active_user, get_current_admin_user
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class UpdateProfileRequest(BaseModel):
|
||||
"""프로필 업데이트 요청"""
|
||||
full_name: str = None
|
||||
theme: str = None
|
||||
language: str = None
|
||||
timezone: str = None
|
||||
|
||||
|
||||
class UpdateUserRequest(BaseModel):
|
||||
"""사용자 정보 업데이트 요청 (관리자용)"""
|
||||
full_name: str = None
|
||||
is_active: bool = None
|
||||
is_admin: bool = None
|
||||
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/profile", response_model=UserInfo)
|
||||
async def get_profile(
|
||||
class UserResponse(BaseModel):
|
||||
"""사용자 응답"""
|
||||
id: str
|
||||
email: str
|
||||
full_name: Optional[str]
|
||||
is_active: bool
|
||||
is_admin: bool
|
||||
role: str
|
||||
can_manage_books: bool
|
||||
can_manage_notes: bool
|
||||
can_manage_novels: bool
|
||||
session_timeout_minutes: int
|
||||
theme: str
|
||||
language: str
|
||||
timezone: str
|
||||
created_at: datetime
|
||||
updated_at: Optional[datetime]
|
||||
last_login: Optional[datetime]
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class CreateUserRequest(BaseModel):
|
||||
"""사용자 생성 요청"""
|
||||
email: EmailStr
|
||||
password: str
|
||||
full_name: Optional[str] = None
|
||||
role: str = "user"
|
||||
can_manage_books: bool = True
|
||||
can_manage_notes: bool = True
|
||||
can_manage_novels: bool = True
|
||||
session_timeout_minutes: int = 5
|
||||
|
||||
|
||||
class UpdateUserRequest(BaseModel):
|
||||
"""사용자 업데이트 요청"""
|
||||
full_name: Optional[str] = None
|
||||
is_active: Optional[bool] = None
|
||||
role: Optional[str] = None
|
||||
can_manage_books: Optional[bool] = None
|
||||
can_manage_notes: Optional[bool] = None
|
||||
can_manage_novels: Optional[bool] = None
|
||||
session_timeout_minutes: Optional[int] = None
|
||||
|
||||
|
||||
class UpdateProfileRequest(BaseModel):
|
||||
"""프로필 업데이트 요청"""
|
||||
full_name: Optional[str] = None
|
||||
theme: Optional[str] = None
|
||||
language: Optional[str] = None
|
||||
timezone: Optional[str] = None
|
||||
|
||||
|
||||
class ChangePasswordRequest(BaseModel):
|
||||
"""비밀번호 변경 요청"""
|
||||
current_password: str
|
||||
new_password: str
|
||||
|
||||
|
||||
@router.get("/me", response_model=UserResponse)
|
||||
async def get_current_user_profile(
|
||||
current_user: User = Depends(get_current_active_user)
|
||||
):
|
||||
"""현재 사용자 프로필 조회"""
|
||||
return UserInfo.from_orm(current_user)
|
||||
return UserResponse(
|
||||
id=str(current_user.id),
|
||||
email=current_user.email,
|
||||
full_name=current_user.full_name,
|
||||
is_active=current_user.is_active,
|
||||
is_admin=current_user.is_admin,
|
||||
role=current_user.role,
|
||||
can_manage_books=current_user.can_manage_books,
|
||||
can_manage_notes=current_user.can_manage_notes,
|
||||
can_manage_novels=current_user.can_manage_novels,
|
||||
session_timeout_minutes=current_user.session_timeout_minutes,
|
||||
theme=current_user.theme,
|
||||
language=current_user.language,
|
||||
timezone=current_user.timezone,
|
||||
created_at=current_user.created_at,
|
||||
updated_at=current_user.updated_at,
|
||||
last_login=current_user.last_login
|
||||
)
|
||||
|
||||
|
||||
@router.put("/profile", response_model=UserInfo)
|
||||
async def update_profile(
|
||||
@router.put("/me", response_model=UserResponse)
|
||||
async def update_current_user_profile(
|
||||
profile_data: UpdateProfileRequest,
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""프로필 업데이트"""
|
||||
update_data = {}
|
||||
"""현재 사용자 프로필 업데이트"""
|
||||
update_fields = profile_data.model_dump(exclude_unset=True)
|
||||
|
||||
if profile_data.full_name is not None:
|
||||
update_data["full_name"] = profile_data.full_name
|
||||
if profile_data.theme is not None:
|
||||
update_data["theme"] = profile_data.theme
|
||||
if profile_data.language is not None:
|
||||
update_data["language"] = profile_data.language
|
||||
if profile_data.timezone is not None:
|
||||
update_data["timezone"] = profile_data.timezone
|
||||
for field, value in update_fields.items():
|
||||
setattr(current_user, field, value)
|
||||
|
||||
if update_data:
|
||||
await db.execute(
|
||||
update(User)
|
||||
.where(User.id == current_user.id)
|
||||
.values(**update_data)
|
||||
current_user.updated_at = datetime.utcnow()
|
||||
await db.commit()
|
||||
await db.refresh(current_user)
|
||||
|
||||
return UserResponse(
|
||||
id=str(current_user.id),
|
||||
email=current_user.email,
|
||||
full_name=current_user.full_name,
|
||||
is_active=current_user.is_active,
|
||||
is_admin=current_user.is_admin,
|
||||
role=current_user.role,
|
||||
can_manage_books=current_user.can_manage_books,
|
||||
can_manage_notes=current_user.can_manage_notes,
|
||||
can_manage_novels=current_user.can_manage_novels,
|
||||
session_timeout_minutes=current_user.session_timeout_minutes,
|
||||
theme=current_user.theme,
|
||||
language=current_user.language,
|
||||
timezone=current_user.timezone,
|
||||
created_at=current_user.created_at,
|
||||
updated_at=current_user.updated_at,
|
||||
last_login=current_user.last_login
|
||||
)
|
||||
|
||||
|
||||
@router.post("/me/change-password")
|
||||
async def change_current_user_password(
|
||||
password_data: ChangePasswordRequest,
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""현재 사용자 비밀번호 변경"""
|
||||
# 현재 비밀번호 확인
|
||||
if not verify_password(password_data.current_password, current_user.hashed_password):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Current password is incorrect"
|
||||
)
|
||||
await db.commit()
|
||||
await db.refresh(current_user)
|
||||
|
||||
return UserInfo.from_orm(current_user)
|
||||
# 새 비밀번호 설정
|
||||
current_user.hashed_password = get_password_hash(password_data.new_password)
|
||||
current_user.updated_at = datetime.utcnow()
|
||||
|
||||
await db.commit()
|
||||
|
||||
return {"message": "Password changed successfully"}
|
||||
|
||||
|
||||
@router.get("/", response_model=List[UserInfo])
|
||||
@router.get("/", response_model=List[UserResponse])
|
||||
async def list_users(
|
||||
admin_user: User = Depends(get_current_admin_user),
|
||||
skip: int = 0,
|
||||
limit: int = 50,
|
||||
current_user: User = Depends(get_current_admin_user),
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""사용자 목록 조회 (관리자 전용)"""
|
||||
result = await db.execute(select(User).order_by(User.created_at.desc()))
|
||||
result = await db.execute(
|
||||
select(User)
|
||||
.order_by(User.created_at.desc())
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
)
|
||||
users = result.scalars().all()
|
||||
return [UserInfo.from_orm(user) for user in users]
|
||||
|
||||
return [
|
||||
UserResponse(
|
||||
id=str(user.id),
|
||||
email=user.email,
|
||||
full_name=user.full_name,
|
||||
is_active=user.is_active,
|
||||
is_admin=user.is_admin,
|
||||
role=user.role,
|
||||
can_manage_books=user.can_manage_books,
|
||||
can_manage_notes=user.can_manage_notes,
|
||||
can_manage_novels=user.can_manage_novels,
|
||||
session_timeout_minutes=user.session_timeout_minutes,
|
||||
theme=user.theme,
|
||||
language=user.language,
|
||||
timezone=user.timezone,
|
||||
created_at=user.created_at,
|
||||
updated_at=user.updated_at,
|
||||
last_login=user.last_login
|
||||
)
|
||||
for user in users
|
||||
]
|
||||
|
||||
|
||||
@router.get("/{user_id}", response_model=UserInfo)
|
||||
async def get_user(
|
||||
user_id: str,
|
||||
admin_user: User = Depends(get_current_admin_user),
|
||||
@router.post("/", response_model=UserResponse)
|
||||
async def create_user(
|
||||
user_data: CreateUserRequest,
|
||||
current_user: User = Depends(get_current_admin_user),
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""특정 사용자 조회 (관리자 전용)"""
|
||||
"""사용자 생성 (관리자 전용)"""
|
||||
# 이메일 중복 확인
|
||||
existing_user = await db.execute(
|
||||
select(User).where(User.email == user_data.email)
|
||||
)
|
||||
if existing_user.scalar_one_or_none():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Email already registered"
|
||||
)
|
||||
|
||||
# 권한 확인 (root만 admin/root 계정 생성 가능)
|
||||
if user_data.role in ["admin", "root"] and current_user.role != "root":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only root users can create admin accounts"
|
||||
)
|
||||
|
||||
# 사용자 생성
|
||||
hashed_password = get_password_hash(user_data.password)
|
||||
|
||||
new_user = User(
|
||||
email=user_data.email,
|
||||
hashed_password=hashed_password,
|
||||
full_name=user_data.full_name,
|
||||
is_active=True,
|
||||
is_admin=user_data.role in ["admin", "root"],
|
||||
role=user_data.role,
|
||||
can_manage_books=user_data.can_manage_books,
|
||||
can_manage_notes=user_data.can_manage_notes,
|
||||
can_manage_novels=user_data.can_manage_novels,
|
||||
session_timeout_minutes=user_data.session_timeout_minutes
|
||||
)
|
||||
|
||||
db.add(new_user)
|
||||
await db.commit()
|
||||
await db.refresh(new_user)
|
||||
|
||||
return UserResponse(
|
||||
id=str(new_user.id),
|
||||
email=new_user.email,
|
||||
full_name=new_user.full_name,
|
||||
is_active=new_user.is_active,
|
||||
is_admin=new_user.is_admin,
|
||||
role=new_user.role,
|
||||
can_manage_books=new_user.can_manage_books,
|
||||
can_manage_notes=new_user.can_manage_notes,
|
||||
can_manage_novels=new_user.can_manage_novels,
|
||||
session_timeout_minutes=new_user.session_timeout_minutes,
|
||||
theme=new_user.theme,
|
||||
language=new_user.language,
|
||||
timezone=new_user.timezone,
|
||||
created_at=new_user.created_at,
|
||||
updated_at=new_user.updated_at,
|
||||
last_login=new_user.last_login
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{user_id}", response_model=UserResponse)
|
||||
async def get_user(
|
||||
user_id: str,
|
||||
current_user: User = Depends(get_current_admin_user),
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""사용자 상세 조회 (관리자 전용)"""
|
||||
result = await db.execute(select(User).where(User.id == user_id))
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
@@ -96,18 +280,34 @@ async def get_user(
|
||||
detail="User not found"
|
||||
)
|
||||
|
||||
return UserInfo.from_orm(user)
|
||||
return UserResponse(
|
||||
id=str(user.id),
|
||||
email=user.email,
|
||||
full_name=user.full_name,
|
||||
is_active=user.is_active,
|
||||
is_admin=user.is_admin,
|
||||
role=user.role,
|
||||
can_manage_books=user.can_manage_books,
|
||||
can_manage_notes=user.can_manage_notes,
|
||||
can_manage_novels=user.can_manage_novels,
|
||||
session_timeout_minutes=user.session_timeout_minutes,
|
||||
theme=user.theme,
|
||||
language=user.language,
|
||||
timezone=user.timezone,
|
||||
created_at=user.created_at,
|
||||
updated_at=user.updated_at,
|
||||
last_login=user.last_login
|
||||
)
|
||||
|
||||
|
||||
@router.put("/{user_id}", response_model=UserInfo)
|
||||
@router.put("/{user_id}", response_model=UserResponse)
|
||||
async def update_user(
|
||||
user_id: str,
|
||||
user_data: UpdateUserRequest,
|
||||
admin_user: User = Depends(get_current_admin_user),
|
||||
current_user: User = Depends(get_current_admin_user),
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""사용자 정보 업데이트 (관리자 전용)"""
|
||||
# 사용자 존재 확인
|
||||
result = await db.execute(select(User).where(User.id == user_id))
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
@@ -117,42 +317,55 @@ async def update_user(
|
||||
detail="User not found"
|
||||
)
|
||||
|
||||
# 자기 자신의 관리자 권한은 제거할 수 없음
|
||||
if user.id == admin_user.id and user_data.is_admin is False:
|
||||
# 권한 확인 (root만 admin/root 계정 수정 가능)
|
||||
if user.role in ["admin", "root"] and current_user.role != "root":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Cannot remove admin privileges from yourself"
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only root users can modify admin accounts"
|
||||
)
|
||||
|
||||
# 업데이트할 데이터 준비
|
||||
update_data = {}
|
||||
if user_data.full_name is not None:
|
||||
update_data["full_name"] = user_data.full_name
|
||||
if user_data.is_active is not None:
|
||||
update_data["is_active"] = user_data.is_active
|
||||
if user_data.is_admin is not None:
|
||||
update_data["is_admin"] = user_data.is_admin
|
||||
# 업데이트할 필드들 적용
|
||||
update_fields = user_data.model_dump(exclude_unset=True)
|
||||
|
||||
if update_data:
|
||||
await db.execute(
|
||||
update(User)
|
||||
.where(User.id == user_id)
|
||||
.values(**update_data)
|
||||
)
|
||||
await db.commit()
|
||||
await db.refresh(user)
|
||||
for field, value in update_fields.items():
|
||||
if field == "role":
|
||||
# 역할 변경 시 is_admin도 함께 업데이트
|
||||
setattr(user, field, value)
|
||||
user.is_admin = value in ["admin", "root"]
|
||||
else:
|
||||
setattr(user, field, value)
|
||||
|
||||
return UserInfo.from_orm(user)
|
||||
user.updated_at = datetime.utcnow()
|
||||
await db.commit()
|
||||
await db.refresh(user)
|
||||
|
||||
return UserResponse(
|
||||
id=str(user.id),
|
||||
email=user.email,
|
||||
full_name=user.full_name,
|
||||
is_active=user.is_active,
|
||||
is_admin=user.is_admin,
|
||||
role=user.role,
|
||||
can_manage_books=user.can_manage_books,
|
||||
can_manage_notes=user.can_manage_notes,
|
||||
can_manage_novels=user.can_manage_novels,
|
||||
session_timeout_minutes=user.session_timeout_minutes,
|
||||
theme=user.theme,
|
||||
language=user.language,
|
||||
timezone=user.timezone,
|
||||
created_at=user.created_at,
|
||||
updated_at=user.updated_at,
|
||||
last_login=user.last_login
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/{user_id}")
|
||||
async def delete_user(
|
||||
user_id: str,
|
||||
admin_user: User = Depends(get_current_admin_user),
|
||||
current_user: User = Depends(get_current_admin_user),
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""사용자 삭제 (관리자 전용)"""
|
||||
# 사용자 존재 확인
|
||||
result = await db.execute(select(User).where(User.id == user_id))
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
@@ -162,15 +375,28 @@ async def delete_user(
|
||||
detail="User not found"
|
||||
)
|
||||
|
||||
# 자기 자신은 삭제할 수 없음
|
||||
if user.id == admin_user.id:
|
||||
# 자기 자신 삭제 방지
|
||||
if user.id == current_user.id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Cannot delete yourself"
|
||||
detail="Cannot delete your own account"
|
||||
)
|
||||
|
||||
# root 계정 삭제 방지
|
||||
if user.role == "root":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Cannot delete root account"
|
||||
)
|
||||
|
||||
# 권한 확인 (root만 admin 계정 삭제 가능)
|
||||
if user.role == "admin" and current_user.role != "root":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only root users can delete admin accounts"
|
||||
)
|
||||
|
||||
# 사용자 삭제
|
||||
await db.execute(delete(User).where(User.id == user_id))
|
||||
await db.commit()
|
||||
|
||||
return {"message": "User deleted successfully"}
|
||||
return {"message": "User deleted successfully"}
|
||||
Reference in New Issue
Block a user