feat: права и инвайты
This commit is contained in:
102
src/domain/workspace_permissions.py
Normal file
102
src/domain/workspace_permissions.py
Normal file
@@ -0,0 +1,102 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from uuid import UUID
|
||||
|
||||
from . import WorkspaceAccessDenied
|
||||
from .workspace import (
|
||||
PermissionKey,
|
||||
PermissionScopeType,
|
||||
Workspace,
|
||||
WorkspaceUser,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WorkspacePermissions:
|
||||
global_permissions: set[PermissionKey]
|
||||
scoped_permissions: dict[tuple[PermissionKey, PermissionScopeType], set[UUID]]
|
||||
|
||||
@classmethod
|
||||
def from_membership(cls, membership: WorkspaceUser) -> WorkspacePermissions:
|
||||
global_permissions = {
|
||||
permission.permission
|
||||
for permission in getattr(membership, 'permissions', []) or []
|
||||
}
|
||||
|
||||
scoped_permissions: dict[tuple[PermissionKey, PermissionScopeType], set[UUID]] = {}
|
||||
for scope in getattr(membership, 'permission_scopes', []) or []:
|
||||
key = (scope.permission, scope.scope_type)
|
||||
scoped_permissions.setdefault(key, set()).add(scope.scope_id)
|
||||
|
||||
return cls(global_permissions=global_permissions, scoped_permissions=scoped_permissions)
|
||||
|
||||
def has_global(self, permission: PermissionKey) -> bool:
|
||||
return permission in self.global_permissions or PermissionKey.ADMIN_FULL in self.global_permissions
|
||||
|
||||
def allowed_project_ids(self, permission: PermissionKey) -> set[UUID] | None:
|
||||
if self.has_global(permission):
|
||||
return None
|
||||
|
||||
allowed: set[UUID] = set()
|
||||
for key in (permission, PermissionKey.ADMIN_FULL):
|
||||
project_scope = self.scoped_permissions.get((key, PermissionScopeType.PROJECT))
|
||||
if project_scope:
|
||||
allowed.update(project_scope)
|
||||
|
||||
return allowed
|
||||
|
||||
def has_permission(
|
||||
self,
|
||||
permission: PermissionKey,
|
||||
*,
|
||||
scope_type: PermissionScopeType | None = None,
|
||||
scope_id: UUID | None = None,
|
||||
) -> bool:
|
||||
if self.has_global(permission):
|
||||
return True
|
||||
|
||||
if scope_type == PermissionScopeType.PROJECT and scope_id is not None:
|
||||
allowed = self.allowed_project_ids(permission)
|
||||
if allowed is None:
|
||||
return True
|
||||
return scope_id in allowed
|
||||
|
||||
return False
|
||||
|
||||
def has_any(self, permission: PermissionKey) -> bool:
|
||||
allowed = self.allowed_project_ids(permission)
|
||||
if allowed is None:
|
||||
return True
|
||||
return bool(allowed)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WorkspacePermissionContext:
|
||||
workspace: Workspace
|
||||
membership: WorkspaceUser
|
||||
permissions: WorkspacePermissions
|
||||
|
||||
def allowed_project_ids(self, permission: PermissionKey) -> set[UUID] | None:
|
||||
return self.permissions.allowed_project_ids(permission)
|
||||
|
||||
def ensure_project_permission(self, permission: PermissionKey, project_id: UUID) -> None:
|
||||
if not self.permissions.has_permission(
|
||||
permission,
|
||||
scope_type=PermissionScopeType.PROJECT,
|
||||
scope_id=project_id,
|
||||
):
|
||||
raise WorkspaceAccessDenied(self.workspace.id)
|
||||
|
||||
|
||||
def build_workspace_permission_context(membership: WorkspaceUser) -> WorkspacePermissionContext:
|
||||
if membership.workspace is None:
|
||||
raise ValueError('Workspace relation must be loaded for membership permissions')
|
||||
|
||||
permissions = WorkspacePermissions.from_membership(membership)
|
||||
|
||||
return WorkspacePermissionContext(
|
||||
workspace=membership.workspace,
|
||||
membership=membership,
|
||||
permissions=permissions,
|
||||
)
|
||||
Reference in New Issue
Block a user