Merge pull request #623 from weni-ai/feature/endpoint-get-flows

Endpoint to get flows

Author: lucaslinhares

temba/api/v2/flows/__init__.py

@@ -0,0 +1,79 @@
+from unittest.mock import patch
+from uuid import uuid4
+
+from django.urls import reverse
+
+from temba.flows.models import Flow
+from temba.tests.base import TembaTest
+
+
+class TestInternalFlowsAPIView(TembaTest):
+    def setUp(self):
+        super().setUp()
+        # ensure project has a proj_uuid for lookups
+        self.org.proj_uuid = uuid4()
+        self.org.save(update_fields=["proj_uuid"])
+
+    @patch("temba.api.v2.flows.views.InternalFlowsAPIView.authentication_classes", [])
+    @patch("temba.api.v2.flows.views.InternalFlowsAPIView.permission_classes", [])
+    def test_list_flows_success(self):
+        # create some flows
+        flow1 = Flow.create(self.org, self.user, name="Flow One")
+        Flow.create(self.org, self.user, name="Flow Two")
+        Flow.create(self.org, self.user, name="Flow Three")
+
+        url = f"{reverse('api.v2.internal_flows')}?project_uuid={self.org.proj_uuid}&limit=2"
+        resp = self.client.get(url)
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+
+        # paginated response shape
+        self.assertIn("results", data)
+        self.assertIn("next", data)
+        self.assertIn("previous", data)
+
+        # only uuid and name returned
+        self.assertEqual(len(data["results"][0].keys()), 2)
+        self.assertIn("uuid", data["results"][0])
+        self.assertIn("name", data["results"][0])
+
+        # ordering should be newest first (flow3, flow2, ...)
+        self.assertEqual({r["name"] for r in data["results"]}, {"Flow Three", "Flow Two"})
+        self.assertIsNotNone(data["next"])  # has more pages
+
+        # follow next page
+        next_url = data["next"]
+        resp = self.client.get(next_url)
+        self.assertEqual(resp.status_code, 200)
+        data2 = resp.json()
+        self.assertEqual(len(data2["results"]), 1)
+        self.assertEqual(data2["results"][0]["name"], flow1.name)
+
+    @patch("temba.api.v2.flows.views.InternalFlowsAPIView.authentication_classes", [])
+    @patch("temba.api.v2.flows.views.InternalFlowsAPIView.permission_classes", [])
+    def test_missing_project_uuid(self):
+        resp = self.client.get(reverse("api.v2.internal_flows"))
+        self.assertEqual(resp.status_code, 400)
+        self.assertEqual(resp.json(), {"error": "project_uuid is required"})
+
+    @patch("temba.api.v2.flows.views.InternalFlowsAPIView.authentication_classes", [])
+    @patch("temba.api.v2.flows.views.InternalFlowsAPIView.permission_classes", [])
+    def test_invalid_project_uuid(self):
+        url = f"{reverse('api.v2.internal_flows')}?project_uuid={uuid4()}&limit=1"
+        resp = self.client.get(url)
+        self.assertEqual(resp.status_code, 404)
+        self.assertEqual(resp.json(), {"error": "Project not found"})
+
+    @patch("temba.api.v2.flows.views.InternalFlowsAPIView.authentication_classes", [])
+    @patch("temba.api.v2.flows.views.InternalFlowsAPIView.permission_classes", [])
+    def test_pagination_limit(self):
+        # create more flows than the limit
+        for i in range(12):
+            Flow.create(self.org, self.user, name=f"Flow {i}")
+
+        url = f"{reverse('api.v2.internal_flows')}?project_uuid={self.org.proj_uuid}&limit=5"
+        resp = self.client.get(url)
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+        self.assertEqual(len(data["results"]), 5)
+        self.assertIsNotNone(data.get("next"))

temba/api/v2/flows/tests.py

@@ -4,13 +4,19 @@
 from rest_framework.exceptions import NotFound, PermissionDenied
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.renderers import JSONRenderer
+from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.views import APIView
+from weni.internal.authenticators import InternalOIDCAuthentication
 
 from django.conf import settings
 
 from temba import mailroom
+from temba.api.v2.internals.views import APIViewMixin
+from temba.api.v2.permissions import IsUserInOrg
+from temba.api.v2.views_base import CreatedOnCursorPagination
 from temba.flows.models import Flow
+from temba.orgs.models import Org
 from temba.utils import analytics
 
 
@@ -99,3 +105,30 @@ def post(self, request, flow_uuid, *args, **kwargs):
                 return Response(client.sim_resume(payload))
             except mailroom.MailroomException:
                 return Response(dict(status="error", description="mailroom error"), status=500)
+
+
+class InternalFlowsAPIView(APIViewMixin, APIView):
+    authentication_classes = [InternalOIDCAuthentication]
+    permission_classes = [IsAuthenticated, IsUserInOrg]
+
+    class Pagination(CreatedOnCursorPagination):
+        page_size_query_param = "limit"
+
+    def get(self, request: Request):
+        params = request.query_params
+        project_uuid = params.get("project_uuid")
+
+        if project_uuid is None:
+            return Response(status=400, data={"error": "project_uuid is required"})
+
+        try:
+            org = Org.objects.get(proj_uuid=project_uuid)
+        except Org.DoesNotExist:
+            return Response(status=404, data={"error": "Project not found"})
+
+        queryset = Flow.objects.filter(org=org, is_active=True)
+
+        paginator = self.Pagination()
+        page = paginator.paginate_queryset(queryset, request, view=self)
+        results = [{"uuid": str(flow.uuid), "name": flow.name} for flow in page]
+        return paginator.get_paginated_response(results)

temba/api/v2/flows/views.py

@@ -4,6 +4,7 @@
 from django.urls import include, path
 
 from temba.api.v2.elasticsearch.views import ContactsElasticSearchEndpoint
+from temba.api.v2.flows.views import InternalFlowsAPIView
 from temba.api.v2.projects.views import GetProjectView
 
 from .flows.urls import urlpatterns as flows_urlpatterns
@@ -83,6 +84,7 @@
     url(r"^events$", EventsEndpoint.as_view(), name="api.v2.events"),
     url(r"^filter_templates$", FilterTemplatesEndpoint.as_view(), name="api.v2.filter_templates"),
     url(r"^filter_templates_new$", FilterTemplatesEndpointNew.as_view(), name="api.v2.filter_templates_new"),
+    url(r"^internal_flows$", InternalFlowsAPIView.as_view(), name="api.v2.internal_flows"),
     url(r"^definitions$", DefinitionsEndpoint.as_view(), name="api.v2.definitions"),
     url(r"^fields$", FieldsEndpoint.as_view(), name="api.v2.fields"),
     url(r"^flow_starts$", FlowStartsEndpoint.as_view(), name="api.v2.flow_starts"),