60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.entities.role import RoleEntity
|
|
from app.infrastructure.database.models.roles import Role
|
|
from app.infrastructure.database.repository.role_repository import RolesRepository
|
|
|
|
|
|
class RolesService:
|
|
def __init__(self, db: Session):
|
|
self.repository = RolesRepository(db)
|
|
|
|
def get_all_roles(self) -> List[RoleEntity]:
|
|
roles = self.repository.get_all()
|
|
return [
|
|
RoleEntity(
|
|
id=role.id,
|
|
name=role.name,
|
|
users=[user.id for user in role.users] if role.users else []
|
|
)
|
|
for role in roles
|
|
]
|
|
|
|
def get_role_by_id(self, role_id: int) -> Optional[RoleEntity]:
|
|
role = self.repository.get_by_id(role_id)
|
|
if role:
|
|
return RoleEntity(
|
|
id=role.id,
|
|
name=role.name,
|
|
users=[user.id for user in role.users] if role.users else []
|
|
)
|
|
return None
|
|
|
|
def create_role(self, entity: RoleEntity) -> RoleEntity:
|
|
role_model = Role(
|
|
name=entity.name
|
|
)
|
|
created_role = self.repository.create(role_model)
|
|
return RoleEntity(
|
|
id=created_role.id,
|
|
name=created_role.name,
|
|
users=[]
|
|
)
|
|
|
|
def update_role(self, role_id: int, entity: RoleEntity) -> Optional[RoleEntity]:
|
|
role_model = self.repository.get_by_id(role_id)
|
|
if role_model:
|
|
role_model.name = entity.name
|
|
self.repository.update(role_model)
|
|
return RoleEntity(
|
|
id=role_model.id,
|
|
name=role_model.name,
|
|
users=[user.id for user in role_model.users] if role_model.users else []
|
|
)
|
|
return None
|
|
|
|
def delete_role(self, role_id: int) -> bool:
|
|
return self.repository.delete(role_id) is not None
|