68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
# app/services/user_service.py
|
|
from typing import Optional, List
|
|
from app.models.user import User
|
|
from app.schemas.user import UserCreate, UserUpdate
|
|
from passlib.context import CryptContext
|
|
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
class UserService:
|
|
@staticmethod
|
|
def get_password_hash(password: str) -> str:
|
|
return pwd_context.hash(password)
|
|
|
|
@staticmethod
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
@staticmethod
|
|
async def get_all_users(skip: int = 0, limit: int = 100) -> List[User]:
|
|
return await User.all().offset(skip).limit(limit)
|
|
|
|
@staticmethod
|
|
async def get_user_by_id(user_id: int) -> Optional[User]:
|
|
return await User.filter(id=user_id).first()
|
|
|
|
@staticmethod
|
|
async def get_user_by_username(username: str) -> Optional[User]:
|
|
return await User.filter(username=username).first()
|
|
|
|
@staticmethod
|
|
async def get_user_by_email(email: str) -> Optional[User]:
|
|
return await User.filter(email=email).first()
|
|
|
|
@staticmethod
|
|
async def create_user(user_data: UserCreate) -> User:
|
|
hashed_password = UserService.get_password_hash(user_data.password)
|
|
user = await User.create(
|
|
username=user_data.username,
|
|
email=user_data.email,
|
|
hashed_password=hashed_password
|
|
)
|
|
return user
|
|
|
|
@staticmethod
|
|
async def update_user(user_id: int, user_data: UserUpdate) -> Optional[User]:
|
|
user = await UserService.get_user_by_id(user_id)
|
|
if not user:
|
|
return None
|
|
|
|
update_data = user_data.model_dump(exclude_unset=True)
|
|
|
|
if "password" in update_data:
|
|
update_data["hashed_password"] = UserService.get_password_hash(
|
|
update_data.pop("password")
|
|
)
|
|
|
|
await user.update_from_dict(update_data)
|
|
await user.save()
|
|
return user
|
|
|
|
@staticmethod
|
|
async def delete_user(user_id: int) -> bool:
|
|
user = await UserService.get_user_by_id(user_id)
|
|
if user:
|
|
await user.delete()
|
|
return True
|
|
return False
|