SDK function for query

Author: matt-aitken

packages/core/src/v3/apiClient/index.ts

@@ -30,6 +30,9 @@ import {
   ListScheduleOptions,
   QueueItem,
   QueueTypeName,
+  QueryExecuteRequestBody,
+  QueryExecuteResponseBody,
+  QueryExecuteCSVResponseBody,
   ReplayRunResponse,
   RescheduleRunRequestBody,
   ResetIdempotencyKeyResponse,
@@ -1406,6 +1409,68 @@ export class ApiClient {
     );
   }
 
+  async executeQuery(
+    query: string,
+    options?: {
+      scope?: "environment" | "project" | "organization";
+      period?: string;
+      from?: string;
+      to?: string;
+      format?: "json" | "csv";
+    },
+    requestOptions?: ZodFetchOptions
+  ): Promise<QueryExecuteResponseBody | QueryExecuteCSVResponseBody> {
+    const body = {
+      query,
+      scope: options?.scope ?? "environment",
+      period: options?.period,
+      from: options?.from,
+      to: options?.to,
+      format: options?.format ?? "json",
+    };
+
+    const format = options?.format ?? "json";
+
+    if (format === "csv") {
+      // For CSV, we get a text response
+      const response = await fetch(`${this.baseUrl}/api/v1/query`, {
+        method: "POST",
+        headers: this.#getHeaders(false),
+        body: JSON.stringify(body),
+      });
+
+      if (!response.ok) {
+        const errText = await response.text().catch((e) => (e as Error).message);
+        let errJSON: Object | undefined;
+        try {
+          errJSON = JSON.parse(errText) as Object;
+        } catch {
+          // ignore
+        }
+        const errMessage = errJSON ? undefined : errText;
+        const responseHeaders = Object.fromEntries(response.headers.entries());
+
+        throw ApiError.generate(response.status, errJSON, errMessage, responseHeaders);
+      }
+
+      return await response.text();
+    }
+
+    // For JSON, use zodfetch
+    return zodfetch(
+      z.object({
+        rows: z.array(z.record(z.any())),
+      }),
+      `${this.baseUrl}/api/v1/query`,
+      {
+        method: "POST",
+        headers: this.#getHeaders(false),
+        body: JSON.stringify(body),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
   #getHeaders(spanParentAsLink: boolean, additionalHeaders?: Record<string, string | undefined>) {
     const headers: Record<string, string> = {
       "Content-Type": "application/json",

packages/core/src/v3/schemas/index.ts

@@ -15,3 +15,4 @@ export * from "./webhooks.js";
 export * from "./checkpoints.js";
 export * from "./warmStart.js";
 export * from "./queues.js";
+export * from "./query.js";

packages/core/src/v3/schemas/query.ts

@@ -0,0 +1,29 @@
+import { z } from "zod";
+
+/**
+ * Request body schema for executing a query
+ */
+export const QueryExecuteRequestBody = z.object({
+  query: z.string(),
+  scope: z.enum(["organization", "project", "environment"]).default("environment"),
+  period: z.string().nullish(),
+  from: z.string().nullish(),
+  to: z.string().nullish(),
+  format: z.enum(["json", "csv"]).default("json"),
+});
+
+export type QueryExecuteRequestBody = z.infer<typeof QueryExecuteRequestBody>;
+
+/**
+ * Response body schema for JSON format queries
+ */
+export const QueryExecuteResponseBody = z.object({
+  rows: z.array(z.record(z.any())),
+});
+
+export type QueryExecuteResponseBody = z.infer<typeof QueryExecuteResponseBody>;
+
+/**
+ * Response body type for CSV format queries (returns a string)
+ */
+export type QueryExecuteCSVResponseBody = string;

packages/trigger-sdk/src/v3/index.ts

@@ -17,6 +17,7 @@ export * from "./otel.js";
 export * from "./schemas.js";
 export * from "./heartbeats.js";
 export * from "./streams.js";
+export * from "./query.js";
 export type { Context };
 
 import type { Context } from "./shared.js";

packages/trigger-sdk/src/v3/query.ts

@@ -0,0 +1,139 @@
+import type {
+  ApiRequestOptions,
+  QueryExecuteResponseBody,
+  QueryExecuteCSVResponseBody,
+} from "@trigger.dev/core/v3";
+import { apiClientManager, mergeRequestOptions } from "@trigger.dev/core/v3";
+import { tracer } from "./tracer.js";
+
+export type QueryScope = "environment" | "project" | "organization";
+export type QueryFormat = "json" | "csv";
+
+/**
+ * Options for executing a TSQL query
+ */
+export type QueryOptions<TFormat extends QueryFormat | undefined = QueryFormat | undefined> = {
+  /**
+   * The scope of the query - determines what data is accessible
+   * - "environment": Current environment only (default)
+   * - "project": All environments in the project
+   * - "organization": All projects in the organization
+   *
+   * @default "environment"
+   */
+  scope?: "environment" | "project" | "organization";
+
+  /**
+   * Time period to query (e.g., "7d", "30d", "1h")
+   * Cannot be used with `from` or `to`
+   */
+  period?: string;
+
+  /**
+   * Start of time range (ISO 8601 timestamp)
+   * Must be used with `to`
+   */
+  from?: string;
+
+  /**
+   * End of time range (ISO 8601 timestamp)
+   * Must be used with `from`
+   */
+  to?: string;
+
+  /**
+   * Response format
+   * - "json": Returns structured data (default)
+   * - "csv": Returns CSV string
+   *
+   * @default "json"
+   */
+  format?: TFormat;
+};
+
+/**
+ * Result type that automatically narrows based on the format option
+ * @template TFormat - The format type (json or csv)
+ * @template TRow - The shape of each row in the result set
+ */
+export type QueryResult<
+  TFormat extends QueryFormat | undefined = undefined,
+  TRow extends Record<string, any> = Record<string, any>
+> = TFormat extends "csv"
+  ? QueryExecuteCSVResponseBody
+  : TFormat extends "json"
+  ? { rows: Array<TRow> }
+  : TFormat extends undefined
+  ? { rows: Array<TRow> }
+  : { rows: Array<TRow> } | QueryExecuteCSVResponseBody;
+
+/**
+ * Execute a TSQL query against your Trigger.dev data
+ *
+ * @template TFormat - The format of the response (inferred from options)
+ * @param {string} tsql - The TSQL query string to execute
+ * @param {QueryOptions<TFormat>} [options] - Optional query configuration
+ * @param {ApiRequestOptions} [requestOptions] - Optional API request configuration
+ * @returns A promise that resolves with the query results
+ *
+ * @example
+ * ```typescript
+ * // Basic query with defaults (environment scope, json format)
+ * const result = await query.execute("SELECT * FROM runs LIMIT 10");
+ * console.log(result.rows);
+ *
+ * // Query with custom period
+ * const lastMonth = await query.execute(
+ *   "SELECT COUNT(*) as count FROM runs",
+ *   { period: "30d" }
+ * );
+ *
+ * // Query with custom date range
+ * const januaryRuns = await query.execute(
+ *   "SELECT * FROM runs",
+ *   {
+ *     from: "2025-01-01T00:00:00Z",
+ *     to: "2025-02-01T00:00:00Z"
+ *   }
+ * );
+ *
+ * // Organization-wide query
+ * const orgStats = await query.execute(
+ *   "SELECT project, COUNT(*) as count FROM runs GROUP BY project",
+ *   { scope: "organization", period: "7d" }
+ * );
+ *
+ * // Export as CSV
+ * const csvData = await query.execute(
+ *   "SELECT * FROM runs",
+ *   { format: "csv", period: "7d" }
+ * );
+ * // csvData is a string containing CSV
+ * ```
+ */
+function execute<TFormat extends QueryFormat | undefined = undefined>(
+  tsql: string,
+  options?: QueryOptions<TFormat>,
+  requestOptions?: ApiRequestOptions
+): Promise<QueryResult<TFormat>> {
+  const apiClient = apiClientManager.clientOrThrow();
+
+  const $requestOptions = mergeRequestOptions(
+    {
+      tracer,
+      name: "query.execute()",
+      icon: "sparkles",
+      attributes: {
+        scope: options?.scope ?? "environment",
+        format: options?.format ?? "json",
+      },
+    },
+    requestOptions
+  );
+
+  return apiClient.executeQuery(tsql, options, $requestOptions) as Promise<QueryResult<TFormat>>;
+}
+
+export const query = {
+  execute,
+};