47 lines
1.2 KiB
Python
47 lines
1.2 KiB
Python
import enum
|
|
import typing
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.db.models.app_user import AppUser
|
|
from src.db.models.budget_form import BudgetForm
|
|
from src.db.models.role import UserRoleEnum
|
|
from src.repository.user_repository import UserRepository
|
|
|
|
|
|
class AccessEnum(int, enum.Enum):
|
|
RESTRICT = 0
|
|
VIEW = 1
|
|
EDIT = 2
|
|
|
|
|
|
class AccessControlService:
|
|
def __init__(self, db: AsyncSession, user: AppUser):
|
|
self.db = db
|
|
self.user_repo = UserRepository(db)
|
|
|
|
async def _check_budget_form(
|
|
self,
|
|
user: AppUser,
|
|
budget_form: BudgetForm,
|
|
) -> AccessEnum:
|
|
if user.role_id == UserRoleEnum.ADMIN:
|
|
return AccessEnum.EDIT
|
|
user = await self.user_repo.get(
|
|
user_id=user.id,
|
|
load_orgs=True,
|
|
)
|
|
if budget_form.org_unit_id in [ou.id for ou in user.org_units]:
|
|
return AccessEnum.EDIT
|
|
return AccessEnum.RESTRICT
|
|
|
|
async def check_object_access(
|
|
self,
|
|
user: AppUser,
|
|
object: typing.Any
|
|
) -> AccessEnum:
|
|
if isinstance(object, BudgetForm):
|
|
return await self._check_budget_form(user=user,budget_form=object)
|
|
return AccessEnum.EDIT
|
|
|