users.py•3.79 kB
import uuid
from typing import Optional
from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from datetime import datetime, timezone
from fastapi import HTTPException
from mcp_host.models.users import User
from mcp_host.schemas.users import UserCreate, UserResponse, UserUpdate
import logging
logger = logging.getLogger(__name__)
class UsersService:
@staticmethod
async def list_users(db: AsyncSession):
try:
statement = select(User)
result = await db.execute(statement)
users = result.scalars().all()
return [
UserResponse.model_validate(user, from_attributes=True)
for user in users
]
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@staticmethod
async def create_user(user_create: UserCreate, db: AsyncSession):
try:
statement = select(User).where(User.email == user_create.email)
result = await db.execute(statement)
user = result.scalar_one_or_none()
if user:
raise HTTPException(status_code=400, detail="User already exists")
user = User(
**user_create.model_dump(),
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
db.add(user)
await db.commit()
await db.refresh(user)
return UserResponse.model_validate(user, from_attributes=True)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@staticmethod
async def get_user_by_id(user_id: UUID, db: AsyncSession):
try:
statement = select(User).where(User.id == user_id)
result = await db.execute(statement)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return UserResponse.model_validate(user, from_attributes=True)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@staticmethod
async def update_user(user_id: UUID, user_update: UserUpdate, db: AsyncSession):
try:
statement = select(User).where(User.id == user_id)
result = await db.execute(statement)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(status_code=404, detail="User not found")
for key, value in user_update.model_dump(exclude_unset=True).items():
if value is not None:
setattr(user, key, value)
await db.commit()
await db.refresh(user)
return UserResponse.model_validate(user, from_attributes=True)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@staticmethod
async def delete_user(user_id: UUID, db: AsyncSession):
try:
statement = select(User).where(User.id == user_id)
result = await db.execute(statement)
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()
return {"message": "User deleted successfully"}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))