[442] - Pagination added for logs

ddpui/celeryworkers/tasks.py

@@ -46,6 +46,8 @@
 from ddpui.utils.singletaskprogress import SingleTaskProgress
 from ddpui.ddpairbyte import airbyte_service, airbytehelpers
 from ddpui.ddpprefect.prefect_service import (
+    get_flow_run_graphs,
+    get_flow_run_logs,
     update_dbt_core_block_schema,
     get_dbt_cli_profile_block,
     prefect_get,
@@ -56,6 +58,7 @@
     TASK_DBTCLEAN,
     TASK_DBTDEPS,
     TASK_AIRBYTESYNC,
+    FLOW_RUN_LOGS_OFFSET_LIMIT
 )
 from ddpui.ddpprefect import DBTCLIPROFILE
 from ddpui.core import llm_service
@@ -750,8 +753,8 @@
     log_file_name = ""
     try:
         if type == LogsSummarizationType.DEPLOYMENT:
-            all_task_logs = get_flow_run_logs_v2(flow_run_id)
-            dbt_tasks = [task for task in all_task_logs if task["id"] == task_id]
+            all_task = get_flow_run_graphs(flow_run_id)
+            dbt_tasks = [task for task in all_task if task["id"] == task_id]
             if len(dbt_tasks) == 0:
                 taskprogress.add(
                     {
@@ -762,6 +765,15 @@
                 )
                 return
             task = dbt_tasks[0]
+            task['logs'] = []
+            limit = 0
+            while(True):
+                new_logs_set = get_flow_run_logs(flow_run_id, task_id, limit, offset=FLOW_RUN_LOGS_OFFSET_LIMIT)
+                task['logs']+=new_logs_set['logs']
+                if len(new_logs_set['logs']) == FLOW_RUN_LOGS_OFFSET_LIMIT:
+                    limit+=1
+                else:
+                    break
             logs_text = "\n".join([log["message"] for log in task["logs"]])
             log_file_name = f"{flow_run_id}_{task_id}"
         elif type == LogsSummarizationType.AIRBYTE_SYNC:

ddpui/ddpprefect/prefect_service.py

@@ -610,12 +610,12 @@
     )
     return res
 
-def get_flow_run_graphs(flow_run_id: str) -> dict:  # pragma: no cover
+def get_flow_run_graphs(flow_run_id: str) -> dict:
     """retreive the tasks from a flow-run from prefect"""
     res = prefect_get(
         f"flow_runs/graph/{flow_run_id}",
     )
     return res
 
 
 def get_flow_run(flow_run_id: str) -> dict:

ddpui/utils/constants.py

@@ -44,3 +44,6 @@
 SYSTEM_USER_EMAIL = "System User"
 
 # prefect flow run states
+
+# offset limit for fetching logs
+FLOW_RUN_LOGS_OFFSET_LIMIT = 200