Hello @jakesippy ,

Instead of using any rest APIs, just run the below cell using spark.sql() programatically, and you will get all the info. The query below will depend on the pipeline event logs and will always give you accurate information.

%sql
DROP VIEW IF EXISTS event_log_raw;
CREATE VIEW event_log_raw AS
SELECT * FROM event_log("3a08341f-1acd-4edd-bf93-691c65907436");

WITH last_status_per_update AS (
  SELECT
    origin.pipeline_id,
    origin.pipeline_name,
    origin.update_id,
    FROM_JSON(details, 'struct<update_progress: struct<state: string>>').update_progress.state AS last_state,
    timestamp,
    ROW_NUMBER() OVER (PARTITION BY origin.update_id ORDER BY timestamp DESC) AS rn
  FROM event_log_raw
  WHERE event_type = 'update_progress'
  QUALIFY rn = 1
),
update_durations AS (
  SELECT
    origin.pipeline_id,
    origin.pipeline_name,
    origin.update_id,
    MIN(CASE WHEN event_type = 'create_update' THEN timestamp END) AS start_time,
    COALESCE(
      MAX(CASE WHEN event_type = 'update_progress'
                AND FROM_JSON(details, 'struct<update_progress: struct<state: string>>').update_progress.state
                    IN ('COMPLETED','FAILED','CANCELED')
               THEN timestamp END),
      current_timestamp()
    ) AS end_time
  FROM event_log_raw
  WHERE event_type IN ('create_update','update_progress')
    AND origin.update_id IS NOT NULL
  GROUP BY origin.pipeline_id, origin.pipeline_name, origin.update_id
  HAVING start_time IS NOT NULL
)
SELECT
  d.pipeline_id,
  d.pipeline_name,
  d.update_id,
  d.start_time,
  d.end_time,
  ROUND(TIMESTAMPDIFF(MILLISECOND, d.start_time, d.end_time)/1000) AS duration_seconds,
  s.last_state AS update_state
FROM update_durations d
JOIN last_status_per_update s
  ON d.pipeline_id = s.pipeline_id AND d.update_id = s.update_id

WHERE d.update_id In ("<update_id>")
ORDER BY d.start_time DESC;

Programmatic way (Python) :

from pyspark.sql import Row

def get_dlt_update_duration_seconds(pipeline_id: str, update_id: str) -> Row:
    """
    Returns a Row with: start_time, end_time, duration_seconds, update_state
    for the given pipeline_id & update_id. Raises ValueError if not found.
    """
    def _sq(s: str) -> str:
        return s.replace("'", "''")
    
    # Step 1: Create the temp view
    spark.sql(
        f"""
        CREATE OR REPLACE TEMP VIEW event_log_raw AS
        SELECT * FROM event_log('{_sq(pipeline_id)}')
        """
    )
    
    # Step 2: Run the main query
    sql = f"""
    WITH last_status_per_update AS (
      SELECT
        origin.pipeline_id,
        origin.pipeline_name,
        origin.update_id,
        FROM_JSON(details, 'struct<update_progress: struct<state: string>>').update_progress.state AS last_state,
        timestamp,
        ROW_NUMBER() OVER (PARTITION BY origin.update_id ORDER BY timestamp DESC) AS rn
      FROM event_log_raw
      WHERE event_type = 'update_progress'
      QUALIFY rn = 1
    ),
    update_durations AS (
      SELECT
        origin.pipeline_id,
        origin.pipeline_name,
        origin.update_id,
        MIN(CASE WHEN event_type = 'create_update' THEN timestamp END) AS start_time,
        COALESCE(
          MAX(CASE WHEN event_type = 'update_progress'
                    AND FROM_JSON(details, 'struct<update_progress: struct<state: string>>').update_progress.state
                        IN ('COMPLETED','FAILED','CANCELED')
                   THEN timestamp END),
          current_timestamp()
        ) AS end_time
      FROM event_log_raw
      WHERE event_type IN ('create_update','update_progress')
        AND origin.update_id IS NOT NULL
      GROUP BY origin.pipeline_id, origin.pipeline_name, origin.update_id
      HAVING start_time IS NOT NULL
    )
    SELECT
      d.start_time,
      d.end_time,
      ROUND(TIMESTAMPDIFF(MILLISECOND, d.start_time, d.end_time)/1000) AS duration_seconds,
      s.last_state AS update_state
    FROM update_durations d
    JOIN last_status_per_update s
      ON d.pipeline_id = s.pipeline_id AND d.update_id = s.update_id
    WHERE d.update_id = '{_sq(update_id)}'
    """
    res = spark.sql(sql)
    rows = res.collect()

    if not rows:
        raise ValueError(f"No update found for pipeline_id={pipeline_id} update_id={update_id}")
    return rows[0]

pipeline_id = "3a08341f-1acd-4edd-bf93-691c65907436"
update_id   = "532c3e5f-d757-4a39-94e3-ebd201edfd75"
r = get_dlt_update_duration_seconds(pipeline_id, update_id)
print(f"Pipeline ID: {pipeline_id}, Update ID: {update_id},Duration(s): {r['duration_seconds']}, state: {r['update_state']}, "
      f"start: {r['start_time']}, end: {r['end_time']}")

Let me know if you have any further questions.