File size: 2,798 Bytes
de0f1ef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | import math
from app.core.exceptions import ConflictError, NotFoundError
from app.core.security import get_password_hash
from app.repositories.user import UserRepository
from app.schemas.user import UserCreate, UserListResponse, UserResponse, UserUpdate
class UserService:
def __init__(self, repo: UserRepository):
self.repo = repo
async def list_users(self, page: int, size: int) -> UserListResponse:
users, total = await self.repo.list(
offset=(page - 1) * size,
limit=size,
order_by=None,
)
return UserListResponse(
items=[UserResponse.model_validate(u) for u in users],
total=total,
page=page,
size=size,
pages=math.ceil(total / size) if total else 1,
)
async def create_user(self, data: UserCreate) -> UserResponse:
if await self.repo.get_by_username(data.username):
raise ConflictError("البيانات المدخلة غير صحيحة")
if data.email and await self.repo.get_by_email(data.email):
raise ConflictError("البيانات المدخلة غير صحيحة")
user = await self.repo.create(
username=data.username,
email=data.email,
password_hash=get_password_hash(data.password),
role=data.role,
)
return UserResponse.model_validate(user)
async def get_user(self, user_id: int) -> UserResponse:
user = await self.repo.get(user_id)
if not user:
raise NotFoundError("User")
return UserResponse.model_validate(user)
async def update_user(self, user_id: int, data: UserUpdate) -> UserResponse:
user = await self.repo.get(user_id)
if not user:
raise NotFoundError("User")
updates = data.model_dump(exclude_unset=True)
if "email" in updates and updates["email"] and updates["email"] != user.email:
existing = await self.repo.get_by_email(updates["email"])
if existing:
raise ConflictError("البيانات المدخلة غير صحيحة") # ✅ تم التعديل
if "password" in updates:
updates["password_hash"] = get_password_hash(updates.pop("password"))
user = await self.repo.update(user, **updates)
return UserResponse.model_validate(user)
async def delete_user(self, user_id: int, current_admin_id: int) -> None:
if user_id == current_admin_id:
raise ConflictError("Cannot delete your own account")
user = await self.repo.get(user_id)
if not user:
raise NotFoundError("User")
await self.repo.delete(user) |