Add Pipeline API support (#85)

Author: ivpusic

src/api/client.ts

@@ -25,6 +25,7 @@ import { QueryAPI } from "./query";
 import { TableAPI } from "./table";
 import { CustomAPI } from "./custom";
 import { UsageAPI } from "./usage";
+import { PipelineAPI } from "./pipeline";
 import { deprecationWarning } from "../deprecation";
 
 /// Various states of query execution that are "terminal".
@@ -50,13 +51,16 @@ export class DuneClient {
   custom: CustomAPI;
   /// Usage Interface
   usage: UsageAPI;
+  /// Pipeline Interface
+  pipeline: PipelineAPI;
 
   constructor(apiKey: string) {
     this.exec = new ExecutionAPI(apiKey);
     this.query = new QueryAPI(apiKey);
     this.table = new TableAPI(apiKey);
     this.custom = new CustomAPI(apiKey);
     this.usage = new UsageAPI(apiKey);
+    this.pipeline = new PipelineAPI(apiKey);
   }
 
   /**

src/api/execution.ts

@@ -13,6 +13,8 @@ import {
   GetResultParams,
   validateAndBuildGetResultParams,
   ExecuteSqlParams,
+  PipelineExecutionResponse,
+  PipelineExecutionParams,
 } from "../types";
 import log from "loglevel";
 import { ageInHours, logPrefix } from "../utils";
@@ -69,6 +71,24 @@ export class ExecutionAPI extends Router {
     return response as ExecutionResponse;
   }
 
+  /**
+   * Executes a query pipeline by query ID according to:
+   * https://docs.dune.com/api-reference/executions/endpoint/execute-query-pipeline
+   * @param {number} queryID id of query to execute pipeline for.
+   * @param {PipelineExecutionParams} params including execution performance.
+   * @returns {PipelineExecutionResponse} response containing pipeline execution ID.
+   */
+  async executeQueryPipeline(
+    queryID: number,
+    params: PipelineExecutionParams = {},
+  ): Promise<PipelineExecutionResponse> {
+    const { performance = QueryEngine.Medium } = params;
+
+    return this.post<PipelineExecutionResponse>(`query/${queryID}/pipeline/execute`, {
+      performance,
+    });
+  }
+
   /**
    * Cancels an execution according to:
    * https://docs.dune.com/api-reference/executions/endpoint/cancel-execution

src/api/index.ts

@@ -1,6 +1,7 @@
 export * from "./client";
 export * from "./custom";
 export * from "./execution";
+export * from "./pipeline";
 export * from "./query";
 export * from "./router";
 export * from "./table";

src/api/pipeline.ts

@@ -0,0 +1,16 @@
+import { PipelineStatusResponse } from "../types";
+import { Router } from "./router";
+
+/**
+ * This class implements all the pipeline routes defined in the Dune API Docs.
+ */
+export class PipelineAPI extends Router {
+  /**
+   * Retrieves the status of a pipeline execution by pipeline execution ID.
+   * @param {string} pipelineExecutionId string representing ID of pipeline execution.
+   * @returns {PipelineStatusResponse} status of pipeline execution.
+   */
+  async getPipelineStatus(pipelineExecutionId: string): Promise<PipelineStatusResponse> {
+    return this._get(`pipelines/executions/${pipelineExecutionId}/status`);
+  }
+}

src/index.ts

@@ -1,3 +1,11 @@
-export { DuneClient, QueryAPI, ExecutionAPI, CustomAPI, TableAPI, UsageAPI } from "./api";
+export {
+  DuneClient,
+  QueryAPI,
+  ExecutionAPI,
+  CustomAPI,
+  TableAPI,
+  UsageAPI,
+  PipelineAPI,
+} from "./api";
 export * from "./types";
 export { Paginator } from "./paginator";

src/types/requestArgs.ts

@@ -212,6 +212,10 @@ export interface ExecuteSqlParams {
   performance?: QueryEngine;
 }
 
+export interface PipelineExecutionParams {
+  performance?: QueryEngine;
+}
+
 export interface BaseCRUDParams extends BaseParams {
   /// Description of the query.
   description?: string;

src/types/response.ts

@@ -41,6 +41,22 @@ export interface ExecutionResponse {
   state: ExecutionState;
 }
 
+export interface PipelineExecutionResponse {
+  pipeline_execution_id: string;
+}
+
+export interface PipelineStatusResponse {
+  status: string;
+  node_executions: {
+    id: number;
+    query_execution_status: {
+      status: string;
+      query_id: number;
+      execution_id: string;
+    };
+  }[];
+}
+
 /// Response resturned from query creation request.
 export interface CreateQueryResponse {
   query_id: number;

tests/e2e/pipeline.spec.ts

@@ -0,0 +1,83 @@
+import { PipelineAPI, ExecutionAPI, DuneClient } from "../../src";
+import log from "loglevel";
+import { QueryEngine } from "../../src/types/requestArgs";
+
+log.setLevel("silent", true);
+
+const API_KEY = process.env.DUNE_API_KEY!;
+
+describe("PipelineAPI: pipeline execution and status", () => {
+  let client: DuneClient;
+  let pipelineAPI: PipelineAPI;
+  let executionAPI: ExecutionAPI;
+  let testQueryId: number;
+
+  beforeAll(() => {
+    client = new DuneClient(API_KEY);
+    pipelineAPI = client.pipeline;
+    executionAPI = client.exec;
+    testQueryId = 6090403;
+  });
+
+  beforeEach((done) => {
+    setTimeout(done, 1000);
+  });
+
+  it("executes a query pipeline successfully", async () => {
+    const response = await executionAPI.executeQueryPipeline(testQueryId, {
+      performance: QueryEngine.Medium,
+    });
+
+    expect(response.pipeline_execution_id).toBeDefined();
+    expect(typeof response.pipeline_execution_id).toBe("string");
+    expect(response.pipeline_execution_id.length).toBeGreaterThan(0);
+  });
+
+  it("executes query pipeline with default performance", async () => {
+    const response = await executionAPI.executeQueryPipeline(testQueryId);
+
+    expect(response.pipeline_execution_id).toBeDefined();
+    expect(typeof response.pipeline_execution_id).toBe("string");
+  });
+
+  it("gets pipeline status after execution", async () => {
+    const executionResponse = await executionAPI.executeQueryPipeline(testQueryId, {
+      performance: QueryEngine.Medium,
+    });
+
+    const pipelineExecutionId = executionResponse.pipeline_execution_id;
+    expect(pipelineExecutionId).toBeDefined();
+
+    const statusResponse = await pipelineAPI.getPipelineStatus(pipelineExecutionId);
+
+    expect(statusResponse.status).toBeDefined();
+    expect(statusResponse.node_executions).toBeDefined();
+    expect(Array.isArray(statusResponse.node_executions)).toBe(true);
+    expect(statusResponse.node_executions.length).toBeGreaterThan(0);
+
+    const firstNode = statusResponse.node_executions[0];
+    expect(firstNode.id).toBeDefined();
+    expect(firstNode.query_execution_status).toBeDefined();
+    expect(firstNode.query_execution_status.status).toBeDefined();
+    expect(firstNode.query_execution_status.query_id).toBe(testQueryId);
+  });
+
+  it("verifies pipeline status response structure", async () => {
+    const executionResponse = await executionAPI.executeQueryPipeline(testQueryId);
+    const pipelineExecutionId = executionResponse.pipeline_execution_id;
+
+    const statusResponse = await pipelineAPI.getPipelineStatus(pipelineExecutionId);
+
+    expect(statusResponse).toHaveProperty("status");
+    expect(statusResponse).toHaveProperty("node_executions");
+    expect(typeof statusResponse.status).toBe("string");
+
+    if (statusResponse.node_executions.length > 0) {
+      const node = statusResponse.node_executions[0];
+      expect(node).toHaveProperty("id");
+      expect(node).toHaveProperty("query_execution_status");
+      expect(node.query_execution_status).toHaveProperty("status");
+      expect(node.query_execution_status).toHaveProperty("query_id");
+    }
+  });
+});