UN-2774 [FEAT] Add sharing functionality for ETL/TASK pipelines (#1545)

johnyrahul · claude · Deepak-Kesavan · commit 1fa86ec442b8 · 2025-10-07T08:59:21.000+05:30
* UN-2774 [FEAT] Add sharing functionality for ETL pipelines - Added shared_users and shared_to_org fields to Pipeline model - Implemented IsOwnerOrSharedUser permission using common permission class - Added API endpoints for managing pipeline sharing - Updated frontend to support sharing modal and user selection - Fixed issue with empty user list in sharing modal 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com> * [REFACTOR] Move Django Q import to top of file Move the Django Q import from inside the method to the top-level imports for better code organization and consistency. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com> * [FEAT] Add granular permissions for pipeline operations - Restrict destroy, update operations to owners only - Allow shared users access to other operations - Add typing import for type annotations 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com> * [FEAT] Enhance sharing notification system - Add conditional import for notification plugin with fallback - Map pipeline types to proper resource types (ETL/TASK) - Improve error handling and logging for notification failures - Prevent update operation failure when notifications fail 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com> * Addressing review comments --------- Co-authored-by: Claude <noreply@anthropic.com>
diff --git a/backend/pipeline_v2/migrations/0003_add_sharing_fields_to_pipeline.py b/backend/pipeline_v2/migrations/0003_add_sharing_fields_to_pipeline.py
@@ -0,0 +1,32 @@
+# Generated by Django 4.2.1 on 2025-09-17 13:18
+
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("pipeline_v2", "0002_remove_pipeline_unique_pipeline_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="pipeline",
+            name="shared_to_org",
+            field=models.BooleanField(
+                db_comment="Whether this pipeline is shared with the entire organization",
+                default=False,
+            ),
+        ),
+        migrations.AddField(
+            model_name="pipeline",
+            name="shared_users",
+            field=models.ManyToManyField(
+                blank=True,
+                db_comment="Users with whom this pipeline is shared",
+                related_name="shared_pipelines",
+                to=settings.AUTH_USER_MODEL,
+            ),
+        ),
+    ]
diff --git a/backend/pipeline_v2/models.py b/backend/pipeline_v2/models.py
@@ -3,6 +3,7 @@
 from account_v2.models import User
 from django.conf import settings
 from django.db import models
+from django.db.models import Q
 from utils.models.base_model import BaseModel
 from utils.models.organization_mixin import (
     DefaultOrganizationManagerMixin,
@@ -18,7 +19,16 @@
 
 
 class PipelineModelManager(DefaultOrganizationManagerMixin, models.Manager):
-    pass
+    def for_user(self, user):
+        """Filter pipelines that the user can access:
+        - Pipelines created by the user
+        - Pipelines shared with the user
+        """
+        return self.filter(
+            Q(created_by=user)  # Owned by user
+            | Q(shared_users=user)  # Shared with user
+            # Q(shared_to_org=True) # Org-wide sharing (optional)
+        ).distinct()
 
 
 class Pipeline(DefaultOrganizationMixin, BaseModel):
@@ -91,6 +101,17 @@ class PipelineStatus(models.TextChoices):
         null=True,
         blank=True,
     )
+    # Sharing fields
+    shared_users = models.ManyToManyField(
+        User,
+        related_name="shared_pipelines",
+        blank=True,
+        db_comment="Users with whom this pipeline is shared",
+    )
+    shared_to_org = models.BooleanField(
+        default=False,
+        db_comment="Whether this pipeline is shared with the entire organization",
+    )
 
     # Manager
     objects = PipelineModelManager()
diff --git a/backend/pipeline_v2/serializers/crud.py b/backend/pipeline_v2/serializers/crud.py
@@ -24,6 +24,7 @@
 
 class PipelineSerializer(IntegrityErrorMixin, AuditSerializer):
     api_endpoint = SerializerMethodField()
+    created_by_email = SerializerMethodField()
 
     class Meta:
         model = Pipeline
@@ -195,6 +196,10 @@ def get_api_endpoint(self, instance: Pipeline):
         """
         return instance.api_endpoint
 
+    def get_created_by_email(self, obj):
+        """Get the creator's email address."""
+        return obj.created_by.email if obj.created_by else None
+
     def create(self, validated_data: dict[str, Any]) -> Any:
         # TODO: Deduce pipeline type based on WF?
         validated_data[PK.ACTIVE] = True
diff --git a/backend/pipeline_v2/serializers/sharing.py b/backend/pipeline_v2/serializers/sharing.py
@@ -0,0 +1,37 @@
+"""Serializers for pipeline sharing functionality."""
+
+from account_v2.serializer import UserSerializer
+from pipeline_v2.models import Pipeline
+from rest_framework import serializers
+from rest_framework.serializers import SerializerMethodField
+
+
+class SharedUserListSerializer(serializers.ModelSerializer):
+    """Serializer for returning pipeline with shared user details."""
+
+    shared_users = SerializerMethodField()
+    created_by = SerializerMethodField()
+    created_by_email = SerializerMethodField()
+
+    class Meta:
+        model = Pipeline
+        fields = [
+            "id",
+            "pipeline_name",
+            "shared_users",
+            "shared_to_org",
+            "created_by",
+            "created_by_email",
+        ]
+
+    def get_shared_users(self, obj):
+        """Get list of shared users with their details."""
+        return UserSerializer(obj.shared_users.all(), many=True).data
+
+    def get_created_by(self, obj):
+        """Get the creator's username."""
+        return obj.created_by.username if obj.created_by else None
+
+    def get_created_by_email(self, obj):
+        """Get the creator's email."""
+        return obj.created_by.email if obj.created_by else None
diff --git a/backend/pipeline_v2/urls.py b/backend/pipeline_v2/urls.py
@@ -31,6 +31,12 @@
     }
 )
 
+list_shared_users = PipelineViewSet.as_view(
+    {
+        "get": PipelineViewSet.list_of_shared_users.__name__,
+    }
+)
+
 pipeline_execute = PipelineViewSet.as_view({"post": "execute"})
 
 
@@ -44,6 +50,11 @@
             name=PipelineURL.EXECUTIONS,
         ),
         path("pipeline/execute/", pipeline_execute, name=PipelineURL.EXECUTE),
+        path(
+            "pipeline/<uuid:pk>/users/",
+            list_shared_users,
+            name="pipeline-shared-users",
+        ),
         path(
             "pipeline/api/postman_collection/<uuid:pk>/",
             download_postman_collection,
diff --git a/backend/pipeline_v2/views.py b/backend/pipeline_v2/views.py
@@ -1,5 +1,6 @@
 import json
 import logging
+from typing import Any
 
 from account_v2.custom_exceptions import DuplicateData
 from api_v2.exceptions import NoActiveAPIKeyError
@@ -8,7 +9,7 @@
 from django.db import IntegrityError
 from django.db.models import QuerySet
 from django.http import HttpResponse
-from permissions.permission import IsOwner
+from permissions.permission import IsOwner, IsOwnerOrSharedUser
 from rest_framework import serializers, status, viewsets
 from rest_framework.decorators import action
 from rest_framework.request import Request
@@ -29,18 +30,35 @@
 from pipeline_v2.serializers.execute import (
     PipelineExecuteSerializer as ExecuteSerializer,
 )
+from pipeline_v2.serializers.sharing import SharedUserListSerializer
+
+try:
+    from plugins.notification.constants import ResourceType
+    from plugins.notification.sharing_notification import SharingNotificationService
+
+    NOTIFICATION_PLUGIN_AVAILABLE = True
+    sharing_notification_service = SharingNotificationService()
+except ImportError:
+    NOTIFICATION_PLUGIN_AVAILABLE = False
+    sharing_notification_service = None
 
 logger = logging.getLogger(__name__)
 
 
 class PipelineViewSet(viewsets.ModelViewSet):
     versioning_class = URLPathVersioning
     queryset = Pipeline.objects.all()
-    permission_classes = [IsOwner]
+
+    def get_permissions(self) -> list[Any]:
+        if self.action in ["destroy", "partial_update", "update"]:
+            return [IsOwner()]
+        return [IsOwnerOrSharedUser()]
+
     serializer_class = PipelineSerializer
 
     def get_queryset(self) -> QuerySet:
-        queryset = Pipeline.objects.filter(created_by=self.request.user)
+        # Use for_user manager method to include shared pipelines
+        queryset = Pipeline.objects.for_user(self.request.user)
 
         # Apply type filter if specified
         pipeline_type = self.request.query_params.get(PipelineConstants.TYPE)
@@ -101,6 +119,59 @@ def perform_destroy(self, instance: Pipeline) -> None:
         super().perform_destroy(instance)
         return SchedulerHelper.remove_job(pipeline_to_remove)
 
+    @action(detail=True, methods=["get"], url_path="users", permission_classes=[IsOwner])
+    def list_of_shared_users(self, request: Request, pk: str | None = None) -> Response:
+        """Returns the list of users the pipeline is shared with."""
+        pipeline = self.get_object()
+        serializer = SharedUserListSerializer(pipeline)
+        return Response(serializer.data, status=status.HTTP_200_OK)
+
+    def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
+        """Override to handle sharing notifications."""
+        instance = self.get_object()
+        current_shared_users = set(instance.shared_users.all())
+
+        response = super().partial_update(request, *args, **kwargs)
+
+        if (
+            response.status_code == 200
+            and "shared_users" in request.data
+            and NOTIFICATION_PLUGIN_AVAILABLE
+        ):
+            try:
+                instance.refresh_from_db()
+                new_shared_users = set(instance.shared_users.all())
+                newly_shared_users = new_shared_users - current_shared_users
+
+                if ResourceType.ETL.value == instance.pipeline_type:
+                    resource_type = ResourceType.ETL.value
+                elif ResourceType.TASK.value == instance.pipeline_type:
+                    resource_type = ResourceType.TASK.value
+
+                if newly_shared_users:
+                    # Only send notifications if there are newly shared users
+                    sharing_notification_service.send_sharing_notification(
+                        resource_type=resource_type,
+                        resource_name=instance.pipeline_name,
+                        resource_id=str(instance.id),
+                        shared_by=request.user,
+                        shared_to=list(newly_shared_users),
+                        resource_instance=instance,
+                    )
+
+                    logger.info(
+                        f"Sent sharing notifications for {instance.pipeline_type} "
+                        f"to {len(newly_shared_users)} users"
+                    )
+
+            except Exception as e:
+                # Log error but don't fail the update operation
+                logger.exception(
+                    f"Failed to send sharing notification, continuing update though: {str(e)}"
+                )
+
+        return response
+
     @action(detail=True, methods=["get"])
     def download_postman_collection(
         self, request: Request, pk: str | None = None
diff --git a/frontend/src/components/pipelines-or-deployments/pipeline-service.js b/frontend/src/components/pipelines-or-deployments/pipeline-service.js
@@ -94,6 +94,32 @@ function pipelineService() {
       };
       return axiosPrivate(requestOptions);
     },
+    getSharedUsers: (pipelineId) => {
+      const requestOptions = {
+        method: "GET",
+        url: `${path}/pipeline/${pipelineId}/users/`,
+      };
+      return axiosPrivate(requestOptions);
+    },
+    updateSharing: (pipelineId, sharedUsers, shareWithEveryone = false) => {
+      const requestOptions = {
+        method: "PATCH",
+        url: `${path}/pipeline/${pipelineId}/`,
+        headers: requestHeaders,
+        data: {
+          shared_users: sharedUsers,
+          shared_to_org: shareWithEveryone,
+        },
+      };
+      return axiosPrivate(requestOptions);
+    },
+    getAllUsers: () => {
+      const requestOptions = {
+        method: "GET",
+        url: `${path}/users/`,
+      };
+      return axiosPrivate(requestOptions);
+    },
   };
 }
 
diff --git a/frontend/src/components/pipelines-or-deployments/pipelines/Pipelines.jsx b/frontend/src/components/pipelines-or-deployments/pipelines/Pipelines.jsx
