64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
from datetime import date
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from src.schemas.users import User, UserUpdate, UserWithTasks
|
|
from src.services.base import BaseService
|
|
|
|
|
|
class UserService(BaseService):
|
|
async def get_user_by_filter(self, **filter_by) -> User | None:
|
|
result = await self.session.user.get_one_or_none(**filter_by)
|
|
if result is None:
|
|
return None
|
|
return User.model_validate(result)
|
|
|
|
async def get_user_by_filter_or_raise(self, **filter_by) -> User:
|
|
user = await self.get_user_by_filter(**filter_by)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return user
|
|
|
|
async def validate_admin_user(self, username: str) -> User:
|
|
user = await self.get_user_by_filter_or_raise(username=username)
|
|
if not user.is_superuser:
|
|
raise HTTPException(status_code=403, detail="Admin access required")
|
|
return user
|
|
|
|
async def get_all_users(self) -> list[User]:
|
|
users = await self.session.user.get_all()
|
|
return [User.model_validate(user) for user in users]
|
|
|
|
async def delete_user(self, id: int) -> None:
|
|
await self.session.user.delete_one(id=id)
|
|
await self.session.commit()
|
|
|
|
async def update_user(self, id: int, update_data: UserUpdate) -> User:
|
|
await self.get_user_by_filter_or_raise(id=id)
|
|
user = await self.session.user.update_one(
|
|
id=id, data=update_data.model_dump(exclude_unset=True)
|
|
)
|
|
await self.session.commit()
|
|
return User.model_validate(user)
|
|
|
|
async def get_user_with_tasks(
|
|
self,
|
|
user_id: int,
|
|
status: str | None,
|
|
limit: int | None,
|
|
offset: int | None,
|
|
date_to: date | None,
|
|
date_from: date | None,
|
|
):
|
|
user = await self.session.user.get_one_with_load(
|
|
user_id=user_id,
|
|
status=status,
|
|
tasks_limit=limit,
|
|
tasks_offset=offset,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found.")
|
|
return UserWithTasks.model_validate(user)
|