get pipeline history for v1

ddpui/api/client/dashboard_api.py

@@ -9,8 +9,9 @@
 from ddpui import auth
 
 # models
-from ddpui.models.org import OrgDataFlow
+from ddpui.models.org import OrgDataFlow, OrgDataFlowv1
 from ddpui.models.orgjobs import BlockLock, DataflowBlock
+from ddpui.models.tasks import DataflowOrgTask, TaskLock
 
 
 dashboardapi = NinjaAPI(urls_namespace="dashboard")
@@ -86,3 +87,47 @@ def get_dashboard(request):
     # we might add more stuff here , system logs etc.
 
     return res
+
+
+@dashboardapi.get("/v1", auth=auth.CanManagePipelines())
+def get_dashboard_v1(request):
+    """Fetch all flows/pipelines created in an organization"""
+    orguser = request.orguser
+
+    org_data_flows = OrgDataFlowv1.objects.filter(
+        org=orguser.org, dataflow_type="orchestrate"
+    ).all()
+
+    res = []
+
+    # fetch 50 (default limit) flow runs for each flow
+    for flow in org_data_flows:
+        task_ids = DataflowOrgTask.objects.filter(dataflow=flow).values(
+            "orgtask__task__id"
+        )
+        # if there is one there will typically be several - a sync,
+        # a git-run, a git-test... we return the userinfo only for the first one
+        lock = TaskLock.objects.filter(
+            orgtask__task__id__in=[x["orgtask__task__id"] for x in task_ids]
+        ).first()
+        res.append(
+            {
+                "name": flow.name,
+                "deploymentId": flow.deployment_id,
+                "cron": flow.cron,
+                "deploymentName": flow.deployment_name,
+                "runs": prefect_service.get_flow_runs_by_deployment_id(
+                    flow.deployment_id, 50
+                ),
+                "lock": {
+                    "lockedBy": lock.locked_by.user.email,
+                    "lockedAt": lock.locked_at,
+                }
+                if lock
+                else None,
+            }
+        )
+
+    # we might add more stuff here , system logs etc.
+
+    return res

ddpui/tests/api_tests/test_dashboard_api.py

@@ -11,7 +11,8 @@
 from ddpui.models.org import Org, OrgPrefectBlock
 from ddpui.models.org_user import User, OrgUser
 from ddpui.models.orgjobs import OrgDataFlow, DataflowBlock, BlockLock
-from ddpui.api.client.dashboard_api import get_dashboard
+from ddpui.models.tasks import Task, DataflowOrgTask, TaskLock, OrgTask, OrgDataFlowv1
+from ddpui.api.client.dashboard_api import get_dashboard, get_dashboard_v1
 
 pytestmark = pytest.mark.django_db
 
@@ -48,3 +49,39 @@ def test_get_dashboard():
     assert result[0]["deploymentName"] == "deployment-name"
     assert result[0]["runs"] == []
     assert result[0]["lock"]["lockedBy"] == "email"
+
+
+def test_get_dashboard_v1():
+    user = User.objects.create(email="email", username="username")
+    org = Org.objects.create(name="org", slug="org")
+    orguser = OrgUser.objects.create(user=user, org=org)
+    request = Mock()
+    request.orguser = orguser
+
+    task = Task.objects.create(
+        type="dbt", slug="dbt-clean", label="DBT clean", command="clean"
+    )
+    orgtask = OrgTask.objects.create(org=org, task=task)
+    odf = OrgDataFlowv1.objects.create(
+        org=org,
+        cron="1",
+        name="flow-name",
+        deployment_id="deployment-id",
+        deployment_name="deployment-name",
+        dataflow_type="orchestrate",
+    )
+    DataflowOrgTask.objects.create(dataflow=odf, orgtask=orgtask)
+    TaskLock.objects.create(orgtask=orgtask, locked_by=orguser)
+
+    with patch(
+        "ddpui.api.client.dashboard_api.prefect_service.get_flow_runs_by_deployment_id"
+    ) as mock_get_flow_runs_by_deployment_id:
+        mock_get_flow_runs_by_deployment_id.return_value = []
+        result = get_dashboard_v1(request)
+
+    assert result[0]["name"] == "flow-name"
+    assert result[0]["deploymentId"] == "deployment-id"
+    assert result[0]["cron"] == "1"
+    assert result[0]["deploymentName"] == "deployment-name"
+    assert result[0]["runs"] == []
+    assert result[0]["lock"]["lockedBy"] == "email"