UN-2022 [FEAT] Add co-owner management for Adapters, API Deployments, Connectors, Pipelines, Workflows, Prompt Studio

backend/adapter_processor_v2/migrations/0004_adapterinstance_co_owners.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.2.1 on 2026-02-17 08:24
+
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("adapter_processor_v2", "0003_mark_deprecated_adapters"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="adapterinstance",
+            name="co_owners",
+            field=models.ManyToManyField(
+                blank=True,
+                help_text="Users with full ownership privileges",
+                related_name="co_owned_adapters",
+                to=settings.AUTH_USER_MODEL,
+            ),
+        ),
+    ]

backend/adapter_processor_v2/migrations/0005_backfill_creator_to_co_owners.py

@@ -0,0 +1,21 @@
+from django.db import migrations
+
+
+def backfill_creator_to_co_owners(apps, schema_editor):
+    adapter_instance_model = apps.get_model("adapter_processor_v2", "AdapterInstance")
+    for adapter in adapter_instance_model.objects.filter(created_by__isnull=False):
+        if not adapter.co_owners.filter(id=adapter.created_by_id).exists():
+            adapter.co_owners.add(adapter.created_by)
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("adapter_processor_v2", "0004_adapterinstance_co_owners"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            backfill_creator_to_co_owners,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

backend/adapter_processor_v2/models.py

@@ -37,7 +37,7 @@ def for_user(self, user: User) -> QuerySet[Any]:
         return (
             self.get_queryset()
             .filter(
-                models.Q(created_by=user)
+                models.Q(co_owners=user)
                 | models.Q(shared_users=user)
                 | models.Q(shared_to_org=True)
                 | models.Q(is_friction_less=True)
@@ -131,6 +131,12 @@ class AdapterInstance(DefaultOrganizationMixin, BaseModel):
     # Introduced field to establish M2M relation between users and adapters.
     # This will introduce intermediary table which relates both the models.
     shared_users = models.ManyToManyField(User, related_name="shared_adapters_instance")
+    co_owners = models.ManyToManyField(
+        User,
+        related_name="co_owned_adapters",
+        blank=True,
+        help_text="Users with full ownership privileges",
+    )
     description = models.TextField(blank=True, null=True, default=None)
 
     objects = AdapterInstanceModelManager()

backend/adapter_processor_v2/serializers.py

@@ -4,6 +4,7 @@
 from account_v2.serializer import UserSerializer
 from cryptography.fernet import Fernet
 from django.conf import settings
+from permissions.co_owner_serializers import CoOwnerRepresentationMixin
 from rest_framework import serializers
 from rest_framework.serializers import ModelSerializer
 
@@ -126,7 +127,7 @@ def get_context_window_size(self, obj: AdapterInstance) -> int:
         return obj.get_context_window_size()
 
 
-class AdapterListSerializer(BaseAdapterSerializer):
+class AdapterListSerializer(CoOwnerRepresentationMixin, BaseAdapterSerializer):
     """Inherits BaseAdapterSerializer.
 
     Used for listing adapters
@@ -175,10 +176,10 @@ def to_representation(self, instance: AdapterInstance) -> dict[str, str]:
         if model:
             rep["model"] = model
 
+        request = self.context.get("request")
+        self.add_co_owner_fields(instance, rep, request)
         if instance.is_friction_less:
             rep["created_by_email"] = "Unstract"
-        else:
-            rep["created_by_email"] = instance.created_by.email
 
         return rep
 
@@ -190,6 +191,7 @@ class SharedUserListSerializer(BaseAdapterSerializer):
     """
 
     shared_users = UserSerializer(many=True)
+    co_owners = UserSerializer(many=True, read_only=True)
     created_by = UserSerializer()
 
     class Meta(BaseAdapterSerializer.Meta):
@@ -200,6 +202,7 @@ class Meta(BaseAdapterSerializer.Meta):
             "adapter_name",
             "adapter_type",
             "created_by",
+            "co_owners",
             "shared_users",
             "shared_to_org",
         )  # type: ignore

backend/adapter_processor_v2/urls.py

@@ -25,6 +25,8 @@
 
 adapter_users = AdapterInstanceViewSet.as_view({"get": "list_of_shared_users"})
 adapter_info = AdapterInstanceViewSet.as_view({"get": "adapter_info"})
+adapter_add_owner = AdapterInstanceViewSet.as_view({"post": "add_co_owner"})
+adapter_remove_owner = AdapterInstanceViewSet.as_view({"delete": "remove_co_owner"})
 urlpatterns = format_suffix_patterns(
     [
         path("adapter_schema/", adapter_schema, name="get_adapter_schema"),
@@ -39,5 +41,15 @@
             adapter_users,
             name="adapter-users",
         ),
+        path(
+            "adapter/<uuid:pk>/owners/",
+            adapter_add_owner,
+            name="adapter-add-owner",
+        ),
+        path(
+            "adapter/<uuid:pk>/owners/<int:user_id>/",
+            adapter_remove_owner,
+            name="adapter-remove-owner",
+        ),
     ]
 )

backend/adapter_processor_v2/views.py

@@ -6,6 +6,7 @@
 from django.db.models import ProtectedError, QuerySet
 from django.http import HttpRequest
 from django.http.response import HttpResponse
+from permissions.co_owner_views import CoOwnerManagementMixin
 from permissions.permission import (
     IsFrictionLessAdapter,
     IsFrictionLessAdapterDelete,
@@ -135,7 +136,7 @@ def test(self, request: Request) -> Response:
         )
 
 
-class AdapterInstanceViewSet(ModelViewSet):
+class AdapterInstanceViewSet(CoOwnerManagementMixin, ModelViewSet):
     serializer_class = AdapterInstanceSerializer
 
     def get_permissions(self) -> list[Any]:
@@ -148,6 +149,9 @@ def get_permissions(self) -> list[Any]:
         elif self.action in ["list_of_shared_users", "adapter_info"]:
             return [IsOwnerOrSharedUserOrSharedToOrg()]
 
+        elif self.action in ["add_co_owner", "remove_co_owner"]:
+            return [IsOwner()]
+
         # Hack for friction-less onboarding
         # User cant view/update metadata but can delete/share etc
         return [IsOwner()]
@@ -198,6 +202,8 @@ def create(self, request: Any) -> Response:
                 )
 
             instance = serializer.save()
+            if instance.created_by:
+                instance.co_owners.add(instance.created_by)
             organization_member = OrganizationMemberService.get_user_by_id(
                 request.user.id
             )

backend/api_v2/api_deployment_views.py

@@ -6,6 +6,7 @@
 from configuration.models import Configuration
 from django.db.models import QuerySet
 from django.http import HttpResponse
+from permissions.co_owner_views import CoOwnerManagementMixin
 from permissions.permission import IsOwner, IsOwnerOrSharedUserOrSharedToOrg
 from plugins import get_plugin
 from prompt_studio.prompt_studio_registry_v2.models import PromptStudioRegistry
@@ -189,9 +190,15 @@ def get(
         )
 
 
-class APIDeploymentViewSet(viewsets.ModelViewSet):
+class APIDeploymentViewSet(CoOwnerManagementMixin, viewsets.ModelViewSet):
     def get_permissions(self) -> list[Any]:
-        if self.action in ["destroy", "partial_update", "update"]:
+        if self.action in [
+            "destroy",
+            "partial_update",
+            "update",
+            "add_co_owner",
+            "remove_co_owner",
+        ]:
             return [IsOwner()]
         return [IsOwnerOrSharedUserOrSharedToOrg()]
 
@@ -260,7 +267,9 @@ def by_prompt_studio_tool(self, request: Request) -> Response:
                 workflow_id__in=workflow_ids, created_by=request.user
             )
 
-            serializer = APIDeploymentListSerializer(deployments, many=True)
+            serializer = APIDeploymentListSerializer(
+                deployments, many=True, context={"request": request}
+            )
             return Response(serializer.data, status=status.HTTP_200_OK)
 
         except PromptStudioRegistry.DoesNotExist:

backend/api_v2/migrations/0004_apideployment_co_owners.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.2.1 on 2026-02-17 08:24
+
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("api_v2", "0003_add_organization_rate_limit"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="apideployment",
+            name="co_owners",
+            field=models.ManyToManyField(
+                blank=True,
+                help_text="Users with full ownership privileges",
+                related_name="co_owned_api_deployments",
+                to=settings.AUTH_USER_MODEL,
+            ),
+        ),
+    ]

backend/api_v2/migrations/0005_backfill_creator_to_co_owners.py

@@ -0,0 +1,21 @@
+from django.db import migrations
+
+
+def backfill_creator_to_co_owners(apps, schema_editor):
+    api_deployment_model = apps.get_model("api_v2", "APIDeployment")
+    for deployment in api_deployment_model.objects.filter(created_by__isnull=False):
+        if not deployment.co_owners.filter(id=deployment.created_by_id).exists():
+            deployment.co_owners.add(deployment.created_by)
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api_v2", "0004_apideployment_co_owners"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            backfill_creator_to_co_owners,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

backend/api_v2/models.py

@@ -27,14 +27,14 @@
 class APIDeploymentModelManager(DefaultOrganizationManagerMixin, models.Manager):
     def for_user(self, user):
         """Filter API deployments that the user can access:
-        - API deployments created by the user
+        - API deployments co-owned by the user
         - API deployments shared with the user
         - API deployments shared with the entire organization
         """
         from django.db.models import Q
 
         return self.filter(
-            Q(created_by=user)  # Owned by user
+            Q(co_owners=user)  # Co-owned by user
             | Q(shared_users=user)  # Shared with user
             | Q(shared_to_org=True)  # Shared to entire organization
         ).distinct()
@@ -96,6 +96,12 @@ class APIDeployment(DefaultOrganizationMixin, BaseModel):
     shared_users = models.ManyToManyField(
         User, related_name="shared_api_deployments", blank=True
     )
+    co_owners = models.ManyToManyField(
+        User,
+        related_name="co_owned_api_deployments",
+        blank=True,
+        help_text="Users with full ownership privileges",
+    )
     shared_to_org = models.BooleanField(
         default=False,
         db_comment="Whether this API deployment is shared with the entire organization",

backend/api_v2/serializers.py

@@ -6,6 +6,10 @@
 
 from django.apps import apps
 from django.core.validators import RegexValidator
+from permissions.co_owner_serializers import (
+    CoOwnerRepresentationMixin,
+    SharedUserListMixin,
+)
 from pipeline_v2.models import Pipeline
 from prompt_studio.prompt_profile_manager_v2.models import ProfileManager
 from rest_framework.serializers import (
@@ -422,9 +426,11 @@ def validate_execution_id(self, value):
         return str(uuid_obj)
 
 
-class APIDeploymentListSerializer(ModelSerializer):
+class APIDeploymentListSerializer(CoOwnerRepresentationMixin, ModelSerializer):
     workflow_name = CharField(source="workflow.workflow_name", read_only=True)
     created_by_email = SerializerMethodField()
+    co_owners_count = SerializerMethodField()
+    is_owner = SerializerMethodField()
 
     class Meta:
         model = APIDeployment
@@ -439,12 +445,28 @@ class Meta:
             "api_name",
             "created_by",
             "created_by_email",
+            "co_owners_count",
+            "is_owner",
         ]
 
     def get_created_by_email(self, obj):
-        """Get the email of the creator."""
+        """Get the email of the primary owner (first co-owner)."""
+        first_co_owner = obj.co_owners.first()
+        if first_co_owner:
+            return first_co_owner.email
         return obj.created_by.email if obj.created_by else None
 
+    def get_co_owners_count(self, obj):
+        """Get the number of co-owners."""
+        return obj.co_owners.count()
+
+    def get_is_owner(self, obj):
+        """Check if the current user is a co-owner."""
+        request = self.context.get("request")
+        if request and hasattr(request, "user"):
+            return obj.co_owners.filter(pk=request.user.pk).exists()
+        return False
+
 
 class APIKeyListSerializer(ModelSerializer):
     class Meta:
@@ -478,22 +500,20 @@ class APIExecutionResponseSerializer(Serializer):
     result = JSONField()
 
 
-class SharedUserListSerializer(ModelSerializer):
+class SharedUserListSerializer(SharedUserListMixin, ModelSerializer):
     """Serializer for returning API deployment with shared user details."""
 
     shared_users = SerializerMethodField()
+    co_owners = SerializerMethodField()
     created_by = SerializerMethodField()
 
     class Meta:
         model = APIDeployment
-        fields = ["id", "display_name", "shared_users", "shared_to_org", "created_by"]
-
-    def get_shared_users(self, obj):
-        """Return list of shared users with id and email."""
-        return [{"id": user.id, "email": user.email} for user in obj.shared_users.all()]
-
-    def get_created_by(self, obj):
-        """Return creator details."""
-        if obj.created_by:
-            return {"id": obj.created_by.id, "email": obj.created_by.email}
-        return None
+        fields = [
+            "id",
+            "display_name",
+            "shared_users",
+            "co_owners",
+            "shared_to_org",
+            "created_by",
+        ]