Alex Blank
a37a3122f9
- Extract tool_types validation to shared module (validate_compose_yaml, check_port_exposed, validate_required_variables) - Extract _get_user and _get_owned_project to auth/dependencies.py - Create useAsyncData hook and apply to 6 pages - Create extractErrorMessage utility - TypeScript and build pass
149 lines
4.2 KiB
Python
149 lines
4.2 KiB
Python
import uuid
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, UploadFile, status
|
|
from pydantic import BaseModel, ConfigDict
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.auth.dependencies import _get_user, get_current_user_id, get_db_session
|
|
from src.models.user import User
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
UPLOAD_DIR = Path("uploads/avatars")
|
|
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
|
ALLOWED_CONTENT_TYPES = {"image/png", "image/jpeg", "image/jpg"}
|
|
MAX_AVATAR_SIZE = 2 * 1024 * 1024 # 2MB
|
|
|
|
|
|
class UserProfileResponse(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: uuid.UUID
|
|
email: str
|
|
name: str
|
|
avatar_url: str | None
|
|
|
|
|
|
class UserProfileUpdate(BaseModel):
|
|
name: str | None = None
|
|
email: str | None = None
|
|
|
|
|
|
@router.get(
|
|
"/me",
|
|
response_model=UserProfileResponse,
|
|
summary="Get current user profile",
|
|
description="Retrieve the profile of the currently authenticated user.",
|
|
)
|
|
async def get_profile(
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> User:
|
|
"""Get the current user's profile.
|
|
|
|
Args:
|
|
user_id: ID of the authenticated user.
|
|
session: Database session.
|
|
|
|
Returns:
|
|
The user's profile information.
|
|
"""
|
|
return await _get_user(session, user_id)
|
|
|
|
|
|
@router.put(
|
|
"/me",
|
|
response_model=UserProfileResponse,
|
|
summary="Update user profile",
|
|
description="Update the current user's profile information.",
|
|
)
|
|
async def update_profile(
|
|
data: UserProfileUpdate,
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> User:
|
|
"""Update the current user's profile.
|
|
|
|
Args:
|
|
data: Profile update data with optional name and email.
|
|
user_id: ID of the authenticated user.
|
|
session: Database session.
|
|
|
|
Returns:
|
|
The updated user profile.
|
|
"""
|
|
user = await _get_user(session, user_id)
|
|
|
|
if data.name is not None:
|
|
if len(data.name.strip()) == 0:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="name cannot be empty")
|
|
user.name = data.name.strip()
|
|
|
|
if data.email is not None:
|
|
if "@" not in data.email:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid email")
|
|
user.email = data.email.strip()
|
|
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
return user
|
|
|
|
|
|
@router.post(
|
|
"/me/avatar",
|
|
response_model=UserProfileResponse,
|
|
summary="Upload avatar",
|
|
description="Upload a profile avatar image (PNG or JPG, max 2MB).",
|
|
)
|
|
async def upload_avatar(
|
|
file: UploadFile,
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> User:
|
|
"""Upload a profile avatar image.
|
|
|
|
Args:
|
|
file: The image file to upload (PNG or JPG, max 2MB).
|
|
user_id: ID of the authenticated user.
|
|
session: Database session.
|
|
|
|
Returns:
|
|
The updated user profile with new avatar URL.
|
|
"""
|
|
user = await _get_user(session, user_id)
|
|
|
|
if file.content_type not in ALLOWED_CONTENT_TYPES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"invalid file type: {file.content_type}. only png and jpg allowed",
|
|
)
|
|
|
|
content = await file.read()
|
|
if len(content) > MAX_AVATAR_SIZE:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="file too large. max size is 2mb",
|
|
)
|
|
|
|
# Delete old avatar if exists
|
|
if user.avatar_url:
|
|
old_path = UPLOAD_DIR / Path(user.avatar_url).name
|
|
if old_path.exists():
|
|
old_path.unlink()
|
|
|
|
# Save new avatar with UUID filename
|
|
filename_part = file.filename or "avatar.png"
|
|
ext = filename_part.split(".")[-1].lower() if "." in filename_part else "png"
|
|
if ext not in {"png", "jpg", "jpeg"}:
|
|
ext = "png"
|
|
|
|
filename = f"{uuid.uuid4()}.{ext}"
|
|
file_path = UPLOAD_DIR / filename
|
|
file_path.write_bytes(content)
|
|
|
|
user.avatar_url = f"/uploads/avatars/{filename}"
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
return user
|