95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.entities.user import UserEntity
|
|
from app.infrastructure.database.models.users import User
|
|
from app.infrastructure.database.repository.user_repository import UserRepository
|
|
|
|
|
|
class UserService:
|
|
def __init__(self, db: Session):
|
|
self.repository = UserRepository(db)
|
|
|
|
def get_all_users(self) -> List[UserEntity]:
|
|
users = self.repository.get_all()
|
|
return [
|
|
UserEntity(
|
|
user.id,
|
|
user.first_name,
|
|
user.last_name,
|
|
user.login,
|
|
user.password,
|
|
user.role_id
|
|
) for user in users
|
|
]
|
|
|
|
def get_user_by_id(self, user_id: int) -> Optional[UserEntity]:
|
|
user = self.repository.get_by_id(user_id)
|
|
if user:
|
|
return UserEntity(
|
|
user.id,
|
|
user.first_name,
|
|
user.last_name,
|
|
user.login,
|
|
user.password,
|
|
user.role_id
|
|
)
|
|
|
|
return None
|
|
|
|
def get_user_by_login(self, login: str) -> Optional[UserEntity]:
|
|
user = self.repository.get_by_login(login)
|
|
if user:
|
|
return UserEntity(
|
|
id=user.id,
|
|
first_name=user.first_name,
|
|
last_name=user.last_name,
|
|
login=user.login,
|
|
password=user.password,
|
|
role_id=user.role_id
|
|
)
|
|
|
|
return None
|
|
|
|
def create_user(self, entity: UserEntity) -> UserEntity:
|
|
user_model = User(
|
|
first_name=entity.first_name,
|
|
last_name=entity.last_name,
|
|
login=entity.login,
|
|
password=entity.password,
|
|
role_id=entity.role_id,
|
|
)
|
|
created_user = self.repository.create(user_model)
|
|
return UserEntity(
|
|
id=created_user.id,
|
|
first_name=created_user.first_name,
|
|
last_name=created_user.last_name,
|
|
login=created_user.login,
|
|
password=created_user.password,
|
|
role_id=created_user.role_id,
|
|
)
|
|
|
|
def update_user(self, user_id: int, entity: UserEntity) -> Optional[UserEntity]:
|
|
user_model = self.repository.get_by_id(user_id)
|
|
if user_model:
|
|
user_model.first_name = entity.first_name
|
|
user_model.last_name = entity.last_name
|
|
user_model.login = entity.login
|
|
user_model.password = entity.password
|
|
user_model.role_id = entity.role_id
|
|
self.repository.update(user_model)
|
|
return UserEntity(
|
|
id=user_model.id,
|
|
first_name=user_model.first_name,
|
|
last_name=user_model.last_name,
|
|
login=user_model.login,
|
|
password=user_model.password,
|
|
role_id=user_model.role_id
|
|
)
|
|
|
|
return None
|
|
|
|
def delete_user(self, user_id: int) -> bool:
|
|
return self.repository.delete(user_id) is not None
|