feat(reports): refine exports and restore project access
This commit is contained in:
144
apps/projects/services/access.py
Normal file
144
apps/projects/services/access.py
Normal file
@@ -0,0 +1,144 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.db.models import Q, QuerySet
|
||||
from django.utils import timezone
|
||||
from rest_framework.exceptions import PermissionDenied, ValidationError
|
||||
|
||||
from apps.projects.models import Project, ProjectAccess
|
||||
from apps.workspaces.models import Workspace, WorkspaceMembership
|
||||
from apps.workspaces.services import PROJECTS_EDIT, get_workspace_role, has_workspace_capability
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
PROJECT_ACCESS_MANAGED_ROLES = {
|
||||
WorkspaceMembership.Role.MEMBER,
|
||||
WorkspaceMembership.Role.GUEST,
|
||||
}
|
||||
PROJECT_ACCESS_IMPLICIT_ROLES = {
|
||||
WorkspaceMembership.Role.OWNER,
|
||||
WorkspaceMembership.Role.ADMIN,
|
||||
}
|
||||
|
||||
|
||||
def user_has_implicit_project_access(user, workspace: Workspace) -> bool:
|
||||
return get_workspace_role(user, workspace) in PROJECT_ACCESS_IMPLICIT_ROLES
|
||||
|
||||
|
||||
def user_has_project_access(user, project: Project) -> bool:
|
||||
if not user or not getattr(user, "is_authenticated", False):
|
||||
return False
|
||||
if user_has_implicit_project_access(user, project.workspace):
|
||||
return True
|
||||
return ProjectAccess.objects.filter(project=project, user=user).exists()
|
||||
|
||||
|
||||
def filter_projects_for_user(user, queryset: QuerySet[Project] | None = None) -> QuerySet[Project]:
|
||||
if queryset is None:
|
||||
queryset = Project.objects.all()
|
||||
|
||||
if not user or not getattr(user, "is_authenticated", False):
|
||||
return queryset.none()
|
||||
|
||||
return queryset.filter(
|
||||
Q(workspace__owner=user)
|
||||
| Q(
|
||||
workspace__memberships__user=user,
|
||||
workspace__memberships__is_active=True,
|
||||
workspace__memberships__role__in=PROJECT_ACCESS_IMPLICIT_ROLES,
|
||||
)
|
||||
| Q(
|
||||
workspace__memberships__user=user,
|
||||
workspace__memberships__is_active=True,
|
||||
workspace__memberships__role__in=PROJECT_ACCESS_MANAGED_ROLES,
|
||||
access_memberships__user=user,
|
||||
)
|
||||
).distinct()
|
||||
|
||||
|
||||
def ensure_project_access(user, project: Project, *, message: str = "Selected project is unavailable.") -> None:
|
||||
if not user_has_project_access(user, project):
|
||||
raise ValidationError({"project_id": message})
|
||||
|
||||
|
||||
def ensure_workspace_project_access(user, workspace: Workspace) -> None:
|
||||
if not has_workspace_capability(user, workspace, PROJECTS_EDIT):
|
||||
raise PermissionDenied("You do not have permission to manage project access in this workspace.")
|
||||
|
||||
|
||||
def get_access_managed_membership(workspace: Workspace, user_id: str) -> WorkspaceMembership:
|
||||
membership = WorkspaceMembership.objects.filter(
|
||||
workspace=workspace,
|
||||
user_id=user_id,
|
||||
is_active=True,
|
||||
is_deleted=False,
|
||||
).select_related("user").first()
|
||||
if not membership:
|
||||
raise ValidationError({"user": "Selected user is not an active member of this workspace."})
|
||||
if membership.role not in PROJECT_ACCESS_MANAGED_ROLES:
|
||||
raise ValidationError({"user": "Owners and admins have implicit access to all projects."})
|
||||
return membership
|
||||
|
||||
|
||||
def build_project_access_items(*, workspace: Workspace, target_user) -> list[dict]:
|
||||
explicit_access_ids = set(
|
||||
ProjectAccess.objects.filter(project__workspace=workspace, user=target_user).values_list("project_id", flat=True)
|
||||
)
|
||||
projects = (
|
||||
Project.objects.filter(workspace=workspace, is_deleted=False)
|
||||
.select_related("client")
|
||||
.order_by("client__name", "name")
|
||||
)
|
||||
return [
|
||||
{
|
||||
"id": str(project.id),
|
||||
"name": project.name,
|
||||
"description": project.description,
|
||||
"color": project.color,
|
||||
"is_archived": project.is_archived,
|
||||
"client": (
|
||||
{"id": str(project.client_id), "name": project.client.name}
|
||||
if project.client_id and project.client
|
||||
else None
|
||||
),
|
||||
"has_access": str(project.id) in {str(project_id) for project_id in explicit_access_ids},
|
||||
}
|
||||
for project in projects
|
||||
]
|
||||
|
||||
|
||||
def grant_project_accesses(*, actor, workspace: Workspace, target_user, project_ids: list[str]) -> int:
|
||||
ensure_workspace_project_access(actor, workspace)
|
||||
get_access_managed_membership(workspace, str(target_user.id))
|
||||
|
||||
projects = list(Project.objects.filter(workspace=workspace, id__in=project_ids, is_deleted=False))
|
||||
if len(projects) != len(set(project_ids)):
|
||||
raise ValidationError({"project_ids": "One or more selected projects do not belong to this workspace."})
|
||||
|
||||
changed = 0
|
||||
for project in projects:
|
||||
access, created, restored = ProjectAccess.get_or_restore(project=project, user=target_user)
|
||||
if created or restored:
|
||||
access.is_active = True
|
||||
access.updated_at = timezone.now()
|
||||
access.save(update_fields=["is_active", "updated_at"])
|
||||
changed += 1
|
||||
return changed
|
||||
|
||||
|
||||
def revoke_project_accesses(*, actor, workspace: Workspace, target_user, project_ids: list[str]) -> int:
|
||||
ensure_workspace_project_access(actor, workspace)
|
||||
get_access_managed_membership(workspace, str(target_user.id))
|
||||
|
||||
accesses = list(
|
||||
ProjectAccess.objects.filter(
|
||||
project__workspace=workspace,
|
||||
user=target_user,
|
||||
project_id__in=project_ids,
|
||||
).select_related("project")
|
||||
)
|
||||
changed = 0
|
||||
for access in accesses:
|
||||
access.delete()
|
||||
changed += 1
|
||||
return changed
|
||||
Reference in New Issue
Block a user