from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from app.database import get_session
from app.models import User, UserCreate, UserRead, UserUpdate
import hashlib
router = APIRouter(prefix="/users", tags=["users"])
def hash_password(password: str) -> str:
"""Simple password hashing for testing purposes."""
return hashlib.sha256(password.encode()).hexdigest()
@router.post("/", response_model=UserRead, status_code=status.HTTP_201_CREATED)
def create_user(user: UserCreate, session: Session = Depends(get_session)):
"""Create a new user."""
# Check if user already exists
existing_user = session.exec(
select(User).where(
(User.username == user.username) | (User.email == user.email)
)
).first()
if existing_user:
if existing_user.username == user.username:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already registered"
)
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
# Create new user
db_user = User(
username=user.username,
email=user.email,
full_name=user.full_name,
is_active=user.is_active,
hashed_password=hash_password(user.password)
)
session.add(db_user)
session.commit()
session.refresh(db_user)
return db_user
@router.get("/", response_model=List[UserRead])
def read_users(skip: int = 0, limit: int = 100, session: Session = Depends(get_session)):
"""Get all users with pagination."""
users = session.exec(select(User).offset(skip).limit(limit)).all()
return users
@router.get("/{user_id}", response_model=UserRead)
def read_user(user_id: int, session: Session = Depends(get_session)):
"""Get user by ID."""
user = session.get(User, user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return user
@router.put("/{user_id}", response_model=UserRead)
def update_user(user_id: int, user_update: UserUpdate, session: Session = Depends(get_session)):
"""Update user by ID."""
user = session.get(User, user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
# Update only provided fields
user_data = user_update.model_dump(exclude_unset=True)
# Handle password update
if "password" in user_data:
user_data["hashed_password"] = hash_password(user_data.pop("password"))
for field, value in user_data.items():
setattr(user, field, value)
session.add(user)
session.commit()
session.refresh(user)
return user
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, session: Session = Depends(get_session)):
"""Delete user by ID."""
user = session.get(User, user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
session.delete(user)
session.commit()
return None
