From fda91cb6d82bf520a8fbe1184608cfcb9a8cb183 Mon Sep 17 00:00:00 2001 From: Artem Tsyrulnikov Date: Mon, 15 Dec 2025 14:10:57 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20=D0=BF=D1=80=D0=B0=D0=B2=D0=B0=20=D1=80?= =?UTF-8?q?=D0=B0=D1=81=D1=88=D0=B8=D1=80=D0=B5=D0=BD=D1=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/domain/workspace.py | 18 +++- src/domain/workspace_permissions.py | 88 +++++++++++++++++-- .../workspace/create_workspace_invite.py | 2 +- 3 files changed, 98 insertions(+), 10 deletions(-) diff --git a/src/domain/workspace.py b/src/domain/workspace.py index 549467d..e3711a7 100644 --- a/src/domain/workspace.py +++ b/src/domain/workspace.py @@ -23,7 +23,7 @@ class Workspace(TimestampedModel): class WorkspaceUserStatus(str, enum.Enum): ACTIVE = 'active' INVITED = 'invited' - REMOVED = 'removed' + BLOCKED = 'blocked' class WorkspaceUser(TimestampedModel): @@ -77,17 +77,29 @@ class WorkspaceInvite(TimestampedModel): class PermissionKey(enum.StrEnum): + ADMIN_FULL = 'admin_full' + PROJECTS_READ = 'projects_read' PROJECTS_WRITE = 'projects_write' + + CREATIVES_READ = 'creatives_read' + CREATIVES_WRITE = 'creatives_write' + PLACEMENTS_READ = 'placements_read' PLACEMENTS_WRITE = 'placements_write' + ANALYTICS_READ = 'analytics_read' - ADMIN_FULL = 'admin_full' + ANALYTICS_WITHOUT_CLICKS = 'analytics_without_clicks' + + CHANNELS_READ = 'channels_read' + CHANNELS_WRITE = 'channels_write' class PermissionScopeType(enum.StrEnum): - WORKSPACE = 'workspace' PROJECT = 'project' + CREATIVE = 'creative' + PLACEMENT = 'placement' + CHANNEL = 'channel' class WorkspaceUserPermission(TimestampedModel): diff --git a/src/domain/workspace_permissions.py b/src/domain/workspace_permissions.py index 9d9211b..6fc3e93 100644 --- a/src/domain/workspace_permissions.py +++ b/src/domain/workspace_permissions.py @@ -19,10 +19,7 @@ class WorkspacePermissions: @classmethod def from_membership(cls, membership: WorkspaceUser) -> WorkspacePermissions: - global_permissions = { - permission.permission - for permission in getattr(membership, 'permissions', []) or [] - } + 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 []: @@ -46,6 +43,42 @@ class WorkspacePermissions: return allowed + def allowed_creative_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): + creative_scope = self.scoped_permissions.get((key, PermissionScopeType.CREATIVE)) + if creative_scope: + allowed.update(creative_scope) + + return allowed + + def allowed_placement_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): + placement_scope = self.scoped_permissions.get((key, PermissionScopeType.PLACEMENT)) + if placement_scope: + allowed.update(placement_scope) + + return allowed + + def allowed_channel_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): + channel_scope = self.scoped_permissions.get((key, PermissionScopeType.CHANNEL)) + if channel_scope: + allowed.update(channel_scope) + + return allowed + def has_permission( self, permission: PermissionKey, @@ -56,8 +89,18 @@ class WorkspacePermissions: 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 scope_type is not None and scope_id is not None: + if scope_type == PermissionScopeType.PROJECT: + allowed = self.allowed_project_ids(permission) + elif scope_type == PermissionScopeType.CREATIVE: + allowed = self.allowed_creative_ids(permission) + elif scope_type == PermissionScopeType.PLACEMENT: + allowed = self.allowed_placement_ids(permission) + elif scope_type == PermissionScopeType.CHANNEL: + allowed = self.allowed_channel_ids(permission) + else: + return False + if allowed is None: return True return scope_id in allowed @@ -80,6 +123,15 @@ class WorkspacePermissionContext: def allowed_project_ids(self, permission: PermissionKey) -> set[UUID] | None: return self.permissions.allowed_project_ids(permission) + def allowed_creative_ids(self, permission: PermissionKey) -> set[UUID] | None: + return self.permissions.allowed_creative_ids(permission) + + def allowed_placement_ids(self, permission: PermissionKey) -> set[UUID] | None: + return self.permissions.allowed_placement_ids(permission) + + def allowed_channel_ids(self, permission: PermissionKey) -> set[UUID] | None: + return self.permissions.allowed_channel_ids(permission) + def ensure_project_permission(self, permission: PermissionKey, project_id: UUID) -> None: if not self.permissions.has_permission( permission, @@ -88,6 +140,30 @@ class WorkspacePermissionContext: ): raise WorkspaceAccessDenied(self.workspace.id) + def ensure_creative_permission(self, permission: PermissionKey, creative_id: UUID) -> None: + if not self.permissions.has_permission( + permission, + scope_type=PermissionScopeType.CREATIVE, + scope_id=creative_id, + ): + raise WorkspaceAccessDenied(self.workspace.id) + + def ensure_placement_permission(self, permission: PermissionKey, placement_id: UUID) -> None: + if not self.permissions.has_permission( + permission, + scope_type=PermissionScopeType.PLACEMENT, + scope_id=placement_id, + ): + raise WorkspaceAccessDenied(self.workspace.id) + + def ensure_channel_permission(self, permission: PermissionKey, channel_id: UUID) -> None: + if not self.permissions.has_permission( + permission, + scope_type=PermissionScopeType.CHANNEL, + scope_id=channel_id, + ): + raise WorkspaceAccessDenied(self.workspace.id) + def build_workspace_permission_context(membership: WorkspaceUser) -> WorkspacePermissionContext: if membership.workspace is None: diff --git a/src/usecase/workspace/create_workspace_invite.py b/src/usecase/workspace/create_workspace_invite.py index 9daae83..3289d55 100644 --- a/src/usecase/workspace/create_workspace_invite.py +++ b/src/usecase/workspace/create_workspace_invite.py @@ -31,7 +31,7 @@ async def create_workspace_invite( raise domain.UserByUsernameNotFound(username) existing_membership = await self.database.get_workspace_membership(workspace_id, invited_user.id) - if existing_membership and existing_membership.status != domain.WorkspaceUserStatus.REMOVED: + if existing_membership and existing_membership.status != domain.WorkspaceUserStatus.BLOCKED: raise domain.WorkspaceMemberAlreadyExists() existing_invite = await self.database.get_workspace_invite_by_user(workspace_id, invited_user.id)