131 lines
3.7 KiB
Python
131 lines
3.7 KiB
Python
"""User management router."""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel, EmailStr
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import get_db
|
|
from app.models.user import User
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# ── Pydantic schemas ──────────────────────────────────────────────────────
|
|
|
|
class UserCreate(BaseModel):
|
|
username: str
|
|
email: EmailStr
|
|
password: str
|
|
|
|
|
|
class UserResponse(BaseModel):
|
|
id: int
|
|
username: str
|
|
email: str
|
|
is_active: bool
|
|
|
|
model_config = {"from_attributes": True}
|
|
|
|
|
|
class UserUpdate(BaseModel):
|
|
email: EmailStr | None = None
|
|
is_active: bool | None = None
|
|
|
|
|
|
# ── Endpoints ─────────────────────────────────────────────────────────────
|
|
|
|
@router.post(
|
|
"/",
|
|
response_model=UserResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
async def create_user(
|
|
user_in: UserCreate,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
):
|
|
"""Register a new user."""
|
|
# Check uniqueness of username
|
|
existing = await db.execute(select(User).where(User.username == user_in.username))
|
|
if existing.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="Username already taken")
|
|
|
|
# Check uniqueness of email
|
|
existing = await db.execute(select(User).where(User.email == user_in.email))
|
|
if existing.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="Email already registered")
|
|
|
|
from app.routers.auth import hash_password
|
|
|
|
hashed = hash_password(user_in.password)
|
|
|
|
user = User(
|
|
username=user_in.username,
|
|
email=user_in.email,
|
|
hashed_password=hashed,
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
@router.get("/me", response_model=UserResponse)
|
|
async def get_current_user(
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
# current_user would come from JWT dependency in real usage
|
|
):
|
|
"""Get the currently authenticated user (placeholder)."""
|
|
raise HTTPException(status_code=501, detail="Not implemented")
|
|
|
|
|
|
@router.get("/{user_id}", response_model=UserResponse)
|
|
async def get_user(
|
|
user_id: int,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
):
|
|
"""Get a user by ID."""
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return user
|
|
|
|
|
|
@router.patch("/{user_id}", response_model=UserResponse)
|
|
async def update_user(
|
|
user_id: int,
|
|
user_update: UserUpdate,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
):
|
|
"""Update a user."""
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
if user_update.email is not None:
|
|
user.email = user_update.email
|
|
if user_update.is_active is not None:
|
|
user.is_active = user_update.is_active
|
|
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_user(
|
|
user_id: int,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
):
|
|
"""Delete a user."""
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
await db.delete(user)
|
|
await db.commit()
|