From 520d3cddddc5b1774bf6f224813e99a415314a60 Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Thu, 8 Oct 2026 16:08:58 +0200 Subject: [PATCH 1/3] feat(appkit): migrate jobs to modular @databricks/sdk-jobs Route the facade's `jobs` accessor to a lazily built JobsClient (Jobs API 2.2) and move the jobs connector onto it: - submit/get/listRuns -> submitRun/getJob/listRunsIter; signal passed as CallOptions, legacy Context adapter removed. - Translate camelCase/bigint models back to the snake_case/number wire shape at the connector boundary, so the plugin's HTTP, SSE and cache output is unchanged (bigint would make JSON.stringify throw). Record fields keep user keys verbatim. - Rebuild runNow's response from the waiter's runId (number_in_job is documented as equal to run_id). - Surface the modular ApiError's httpStatusCode as statusCode so upstream 4xx keep mapping to the right HTTP status. The CLI's jobs calls stay on the legacy client via toLegacyWorkspaceClient. Co-authored-by: Isaac Signed-off-by: MarioCadenas --- knip.json | 1 + packages/appkit/package.json | 1 + packages/appkit/src/connectors/jobs/client.ts | 176 ++++++++++++++---- .../src/connectors/jobs/tests/client.test.ts | 74 ++++++++ .../src/plugins/jobs/tests/plugin.test.ts | 157 ++++++++-------- packages/shared/package.json | 1 + .../shared/src/workspace-client/client.ts | 11 +- .../shared/src/workspace-client/modular.ts | 23 ++- packages/shared/src/workspace-client/types.ts | 10 +- pnpm-lock.yaml | 19 ++ 10 files changed, 350 insertions(+), 123 deletions(-) create mode 100644 packages/appkit/src/connectors/jobs/tests/client.test.ts diff --git a/knip.json b/knip.json index c57c8be24..14df12e42 100644 --- a/knip.json +++ b/knip.json @@ -14,6 +14,7 @@ "@databricks/sdk-core", "@databricks/sdk-experimental", "@databricks/sdk-genie", + "@databricks/sdk-jobs", "@databricks/sdk-options", "@databricks/sdk-scim", "@databricks/sdk-statementexecution", diff --git a/packages/appkit/package.json b/packages/appkit/package.json index 3eda04dba..33899de06 100644 --- a/packages/appkit/package.json +++ b/packages/appkit/package.json @@ -75,6 +75,7 @@ "@databricks/sdk-core": "0.51.0", "@databricks/sdk-experimental": "0.17.0", "@databricks/sdk-genie": "0.54.0", + "@databricks/sdk-jobs": "0.57.0", "@databricks/sdk-options": "0.51.0", "@databricks/sdk-scim": "0.51.0", "@databricks/sdk-statementexecution": "0.52.0", diff --git a/packages/appkit/src/connectors/jobs/client.ts b/packages/appkit/src/connectors/jobs/client.ts index a00474d21..ab048a202 100644 --- a/packages/appkit/src/connectors/jobs/client.ts +++ b/packages/appkit/src/connectors/jobs/client.ts @@ -9,15 +9,95 @@ import { SpanStatusCode, TelemetryManager, } from "../../telemetry"; -import { - Context, - type jobs, - type WorkspaceClient, +import type { + GetJobRequest, + GetRunRequest, + jobs, + ListRunsRequest, + RunNowRequest, + SubmitRunRequest, + WorkspaceClient, } from "../../workspace-client"; import type { JobsConnectorConfig } from "./types"; const logger = createLogger("connectors:jobs"); +/** + * `Record` fields of the Jobs model. Their keys are user data + * (notebook params, tags, Spark conf), so they are copied verbatim instead of + * re-cased. + */ +const MAP_FIELDS = new Set([ + "artifactsHeaders", + "baseParameters", + "customTags", + "filters", + "jobParameters", + "namedParameters", + "notebookBaseParameters", + "notebookParams", + "parameters", + "pipelineTaskParameters", + "pythonNamedParams", + "sparkConf", + "sparkEnvVars", + "sqlParams", + "tags", + "variables", + "violations", +]); + +/** The one model field whose camelCase name doesn't round-trip to its wire key. */ +const WIRE_KEY_OVERRIDES: Record = { + pipelineTaskParameters: "parameters", +}; + +function isPlainObject(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +/** + * Modular model → the legacy wire shape the plugin's public API (HTTP JSON, SSE, + * cache) has always exposed: snake_case keys and `number` int64s. `bigint` would + * make `JSON.stringify` throw. `Number()` loses precision past 2^53, exactly as + * the legacy SDK's plain `JSON.parse` did. + */ +function toWire(value: unknown, verbatimKeys = false): unknown { + if (typeof value === "bigint") return Number(value); + if (Array.isArray(value)) return value.map((v) => toWire(v)); + if (!isPlainObject(value)) return value; + return Object.fromEntries( + Object.entries(value).map(([key, v]) => + verbatimKeys + ? [key, v] + : [ + WIRE_KEY_OVERRIDES[key] ?? + key.replace(/[A-Z]/g, (c) => `_${c.toLowerCase()}`), + toWire(v, MAP_FIELDS.has(key)), + ], + ), + ); +} + +/** + * Legacy snake_case request → modular camelCase request. int64 fields (`job_id`, + * `run_id`) still need an explicit `BigInt()` at the call site: the SDK's + * marshal schemas reject a `number` there. + */ +function fromWire(value: unknown, verbatimKeys = false): unknown { + if (Array.isArray(value)) return value.map((v) => fromWire(v)); + if (!isPlainObject(value)) return value; + return Object.fromEntries( + Object.entries(value).map(([key, v]) => { + if (verbatimKeys) return [key, v]; + const camel = key.replace(/_([a-z0-9])/g, (_, c: string) => + c.toUpperCase(), + ); + return [camel, fromWire(v, MAP_FIELDS.has(camel))]; + }), + ); +} + export class JobsConnector { private readonly name = "jobs"; private readonly config: JobsConnectorConfig; @@ -55,7 +135,11 @@ export class JobsConnector { signal?: AbortSignal, ): Promise { return this._callApi("submit", async () => { - return workspaceClient.jobs.submit(request, this._createContext(signal)); + const waiter = await workspaceClient.jobs.submitRun( + fromWire(request) as SubmitRunRequest, + { signal }, + ); + return { run_id: Number(waiter.runId) }; }); } @@ -65,7 +149,17 @@ export class JobsConnector { signal?: AbortSignal, ): Promise { return this._callApi("runNow", async () => { - return workspaceClient.jobs.runNow(request, this._createContext(signal)); + const waiter = await workspaceClient.jobs.runNow( + { + ...(fromWire(request) as RunNowRequest), + jobId: BigInt(request.job_id), + }, + { signal }, + ); + // The waiter only exposes runId; the API documents number_in_job as + // "set to the same value as run_id". + const runId = Number(waiter.runId); + return { run_id: runId, number_in_job: runId }; }); } @@ -75,7 +169,14 @@ export class JobsConnector { signal?: AbortSignal, ): Promise { return this._callApi("getRun", async () => { - return workspaceClient.jobs.getRun(request, this._createContext(signal)); + const run = await workspaceClient.jobs.getRun( + { + ...(fromWire(request) as GetRunRequest), + runId: BigInt(request.run_id), + }, + { signal }, + ); + return toWire(run) as jobs.Run; }); } @@ -85,10 +186,11 @@ export class JobsConnector { signal?: AbortSignal, ): Promise { return this._callApi("getRunOutput", async () => { - return workspaceClient.jobs.getRunOutput( - request, - this._createContext(signal), + const output = await workspaceClient.jobs.getRunOutput( + { runId: BigInt(request.run_id) }, + { signal }, ); + return toWire(output) as jobs.RunOutput; }); } @@ -98,9 +200,9 @@ export class JobsConnector { signal?: AbortSignal, ): Promise { await this._callApi("cancelRun", async () => { - return workspaceClient.jobs.cancelRun( - request, - this._createContext(signal), + await workspaceClient.jobs.cancelRun( + { runId: BigInt(request.run_id) }, + { signal }, ); }); } @@ -113,11 +215,16 @@ export class JobsConnector { return this._callApi("listRuns", async () => { const runs: jobs.BaseRun[] = []; const limit = Math.max(1, Math.min(request.limit ?? 100, 100)); - for await (const run of workspaceClient.jobs.listRuns( - { ...request, limit }, - this._createContext(signal), + for await (const run of workspaceClient.jobs.listRunsIter( + { + ...(fromWire(request) as ListRunsRequest), + jobId: + request.job_id === undefined ? undefined : BigInt(request.job_id), + limit, + }, + { signal }, )) { - runs.push(run); + runs.push(toWire(run) as jobs.BaseRun); if (runs.length >= limit) break; } return runs; @@ -130,7 +237,14 @@ export class JobsConnector { signal?: AbortSignal, ): Promise { return this._callApi("getJob", async () => { - return workspaceClient.jobs.get(request, this._createContext(signal)); + const job = await workspaceClient.jobs.getJob( + { + ...(fromWire(request) as GetJobRequest), + jobId: BigInt(request.job_id), + }, + { signal }, + ); + return toWire(job) as jobs.Job; }); } @@ -164,6 +278,17 @@ export class JobsConnector { if (error instanceof AppKitError) { throw error; } + // The modular SDK's ApiError exposes the HTTP status as + // `httpStatusCode` (-1 when not an HTTP error); Plugin.execute() + // maps on `statusCode`. + if ( + error instanceof Error && + "httpStatusCode" in error && + typeof error.httpStatusCode === "number" && + error.httpStatusCode > 0 + ) { + throw Object.assign(error, { statusCode: error.httpStatusCode }); + } // Preserve SDK ApiError (and any error with a numeric statusCode) // so Plugin.execute() can map it to the correct HTTP status. if ( @@ -198,19 +323,4 @@ export class JobsConnector { { name: this.name, includePrefix: true }, ); } - - private _createContext(signal?: AbortSignal) { - return new Context({ - cancellationToken: { - // Getter — evaluated on every read so SDK code paths that poll - // (rather than subscribe) observe cancellation live. - get isCancellationRequested() { - return signal?.aborted ?? false; - }, - onCancellationRequested: (cb: () => void) => { - signal?.addEventListener("abort", cb, { once: true }); - }, - }, - }); - } } diff --git a/packages/appkit/src/connectors/jobs/tests/client.test.ts b/packages/appkit/src/connectors/jobs/tests/client.test.ts new file mode 100644 index 000000000..542175284 --- /dev/null +++ b/packages/appkit/src/connectors/jobs/tests/client.test.ts @@ -0,0 +1,74 @@ +import { describe, expect, test, vi } from "vitest"; + +import { JobsConnector } from "../client"; + +// The modular Jobs SDK returns camelCase models with `bigint` int64s, which +// `JSON.stringify` rejects. The connector must hand the plugin the legacy wire +// shape (snake_case, `number`) its public HTTP/SSE/cache surface has always had. +describe("JobsConnector wire-shape translation", () => { + test("getRun returns a JSON-serializable snake_case run with numeric IDs", async () => { + const getRun = vi.fn().mockResolvedValue({ + runId: 9007199254740991n, + jobId: 123n, + startTime: 1700000000000n, + state: { lifeCycleState: "TERMINATED" }, + overridingParameters: { notebookParams: { myParam: "x" } }, + tasks: [ + { + taskKey: "t1", + pipelineTask: { pipelineTaskParameters: { fullRefresh: "true" } }, + }, + ], + }); + const client = { jobs: { getRun } } as never; + + const run = await new JobsConnector({}).getRun(client, { run_id: 42 }); + + expect(getRun).toHaveBeenCalledWith({ runId: 42n }, { signal: undefined }); + expect(JSON.parse(JSON.stringify(run))).toEqual({ + run_id: 9007199254740991, + job_id: 123, + start_time: 1700000000000, + state: { life_cycle_state: "TERMINATED" }, + // Map keys are user data: copied verbatim, not re-cased. + overriding_parameters: { notebook_params: { myParam: "x" } }, + tasks: [ + { + task_key: "t1", + pipeline_task: { parameters: { fullRefresh: "true" } }, + }, + ], + }); + }); + + test("runNow camelCases the request without touching param keys", async () => { + const runNow = vi.fn().mockResolvedValue({ runId: 7n }); + const client = { jobs: { runNow } } as never; + + const result = await new JobsConnector({}).runNow(client, { + job_id: 123, + notebook_params: { my_param: "v" }, + }); + + expect(runNow).toHaveBeenCalledWith( + { jobId: 123n, notebookParams: { my_param: "v" } }, + { signal: undefined }, + ); + expect(result).toEqual({ run_id: 7, number_in_job: 7 }); + }); + + test("surfaces the modular ApiError's httpStatusCode as statusCode", async () => { + // Shaped like sdk-core's ApiError: the status is a getter, not `statusCode`. + class ModularApiError extends Error { + get httpStatusCode() { + return 404; + } + } + const getRun = vi.fn().mockRejectedValue(new ModularApiError("not found")); + const client = { jobs: { getRun } } as never; + + await expect( + new JobsConnector({}).getRun(client, { run_id: 1 }), + ).rejects.toMatchObject({ statusCode: 404, message: "not found" }); + }); +}); diff --git a/packages/appkit/src/plugins/jobs/tests/plugin.test.ts b/packages/appkit/src/plugins/jobs/tests/plugin.test.ts index 84b44ed0c..a9fc1a33d 100644 --- a/packages/appkit/src/plugins/jobs/tests/plugin.test.ts +++ b/packages/appkit/src/plugins/jobs/tests/plugin.test.ts @@ -24,29 +24,22 @@ const { mockClient, jobsApi } = await vi.hoisted(async () => { const mockClient = createMockWorkspaceClient(); - // Facade accessors are typed against the legacy SDK, so `.mockResolvedValue` + // Facade accessors are typed against the SDK, so `.mockResolvedValue` // on them would not typecheck. `getMock` is the typed handle; it mints // idempotently, so these are the very functions the plugin will call. const jobsApi = { runNow: getMock(mockClient, "jobs.runNow"), - submit: getMock(mockClient, "jobs.submit"), + submitRun: getMock(mockClient, "jobs.submitRun"), getRun: getMock(mockClient, "jobs.getRun"), getRunOutput: getMock(mockClient, "jobs.getRunOutput"), cancelRun: getMock(mockClient, "jobs.cancelRun"), - listRuns: getMock(mockClient, "jobs.listRuns"), - get: getMock(mockClient, "jobs.get"), + listRunsIter: getMock(mockClient, "jobs.listRunsIter"), + getJob: getMock(mockClient, "jobs.getJob"), }; return { mockClient, jobsApi }; }); -vi.mock("../../../workspace-client", async (importOriginal) => { - const actual = - await importOriginal(); - // Only `Context` — the client itself is injected through ServiceContext. - return { ...actual, Context: vi.fn() }; -}); - // Boots AppKit's real in-memory cache (no cache-module mock needed). useTestCache(); @@ -264,7 +257,7 @@ describe("JobsPlugin", () => { test("runNow passes configured job_id to connector", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); const plugin = new JobsPlugin({}); const exported = plugin.exports(); @@ -273,7 +266,7 @@ describe("JobsPlugin", () => { await handle.runNow(); expect(jobsApi.runNow).toHaveBeenCalledWith( - expect.objectContaining({ job_id: 123 }), + expect.objectContaining({ jobId: 123n }), expect.anything(), ); }); @@ -281,7 +274,7 @@ describe("JobsPlugin", () => { test("runNow merges user params with configured job_id (no taskType)", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); const plugin = new JobsPlugin({}); const exported = plugin.exports(); @@ -293,8 +286,8 @@ describe("JobsPlugin", () => { expect(jobsApi.runNow).toHaveBeenCalledWith( expect.objectContaining({ - job_id: 123, - notebook_params: { key: "value" }, + jobId: 123n, + notebookParams: { key: "value" }, }), expect.anything(), ); @@ -323,7 +316,7 @@ describe("JobsPlugin", () => { test("runNow maps validated params to SDK fields when taskType is set", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); const plugin = new JobsPlugin({ jobs: { @@ -339,8 +332,8 @@ describe("JobsPlugin", () => { expect(jobsApi.runNow).toHaveBeenCalledWith( expect.objectContaining({ - job_id: 123, - notebook_params: { key: "value" }, + jobId: 123n, + notebookParams: { key: "value" }, }), expect.anything(), ); @@ -349,7 +342,7 @@ describe("JobsPlugin", () => { test("runNow skips validation when no schema is configured", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); const plugin = new JobsPlugin({}); const handle = plugin.exports()("etl"); @@ -363,8 +356,8 @@ describe("JobsPlugin", () => { process.env.DATABRICKS_JOB_ETL = "123"; jobsApi.getRun.mockResolvedValue({ - run_id: 1, - state: { life_cycle_state: "TERMINATED" }, + runId: 1n, + state: { lifeCycleState: "TERMINATED" }, }); const plugin = new JobsPlugin({}); @@ -389,7 +382,7 @@ describe("JobsPlugin", () => { test("getJob wraps call in execute", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.get.mockResolvedValue({ job_id: 123 }); + jobsApi.getJob.mockResolvedValue({ jobId: 123n }); const plugin = new JobsPlugin({}); const executeSpy = vi.spyOn(plugin as any, "execute"); @@ -413,7 +406,7 @@ describe("JobsPlugin", () => { test("listRuns clamps caller-supplied limit before calling the SDK", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.listRuns.mockReturnValue((async function* () {})()); + jobsApi.listRunsIter.mockReturnValue((async function* () {})()); const plugin = new JobsPlugin({}); const handle = plugin.exports()("etl"); @@ -421,7 +414,7 @@ describe("JobsPlugin", () => { await handle.listRuns({ limit: 10000 }); // SDK should receive the clamped limit, not the caller-supplied 10000. - expect(jobsApi.listRuns).toHaveBeenCalledWith( + expect(jobsApi.listRunsIter).toHaveBeenCalledWith( expect.objectContaining({ limit: 100 }), expect.anything(), ); @@ -431,7 +424,7 @@ describe("JobsPlugin", () => { process.env.DATABRICKS_JOB_ETL = "123"; // Pre-flight getRun verifies the run belongs to the configured jobId. - jobsApi.getRun.mockResolvedValue({ run_id: 1, job_id: 123 }); + jobsApi.getRun.mockResolvedValue({ runId: 1n, jobId: 123n }); jobsApi.cancelRun.mockResolvedValue(undefined); const plugin = new JobsPlugin({}); @@ -452,15 +445,15 @@ describe("JobsPlugin", () => { test("runAndWait yields status updates and terminates on TERMINATED", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); jobsApi.getRun .mockResolvedValueOnce({ - run_id: 42, - state: { life_cycle_state: "RUNNING" }, + runId: 42n, + state: { lifeCycleState: "RUNNING" }, }) .mockResolvedValueOnce({ - run_id: 42, - state: { life_cycle_state: "TERMINATED" }, + runId: 42n, + state: { lifeCycleState: "TERMINATED" }, }); const plugin = new JobsPlugin({ pollIntervalMs: 10 }); @@ -544,7 +537,7 @@ describe("JobsPlugin", () => { test("listRuns returns error result on execute failure", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.listRuns.mockImplementation(() => { + jobsApi.listRunsIter.mockImplementation(() => { throw new Error("Auth failure"); }); @@ -587,7 +580,7 @@ describe("JobsPlugin", () => { test("successful operations return ok result with data", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); const plugin = new JobsPlugin({}); const handle = plugin.exports()("etl"); @@ -604,7 +597,7 @@ describe("JobsPlugin", () => { test("getRun returns 404 when run.job_id does not match configured jobId", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.getRun.mockResolvedValue({ run_id: 99, job_id: 456 }); + jobsApi.getRun.mockResolvedValue({ runId: 99n, jobId: 456n }); const plugin = new JobsPlugin({}); const handle = plugin.exports()("etl"); @@ -617,7 +610,7 @@ describe("JobsPlugin", () => { test("getRunOutput returns 404 when run belongs to another job", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.getRun.mockResolvedValue({ run_id: 99, job_id: 456 }); + jobsApi.getRun.mockResolvedValue({ runId: 99n, jobId: 456n }); jobsApi.getRunOutput.mockResolvedValue({ logs: "nope" }); const plugin = new JobsPlugin({}); @@ -633,7 +626,7 @@ describe("JobsPlugin", () => { test("cancelRun returns 404 when run belongs to another job", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.getRun.mockResolvedValue({ run_id: 99, job_id: 456 }); + jobsApi.getRun.mockResolvedValue({ runId: 99n, jobId: 456n }); jobsApi.cancelRun.mockResolvedValue(undefined); const plugin = new JobsPlugin({}); @@ -649,9 +642,9 @@ describe("JobsPlugin", () => { process.env.DATABRICKS_JOB_ETL = "123"; jobsApi.getRun.mockResolvedValue({ - run_id: 42, - job_id: 123, - state: { life_cycle_state: "TERMINATED" }, + runId: 42n, + jobId: 123n, + state: { lifeCycleState: "TERMINATED" }, }); const plugin = new JobsPlugin({}); @@ -662,17 +655,11 @@ describe("JobsPlugin", () => { if (result.ok) expect(result.data.run_id).toBe(42); }); - test("connector's cancellation token reflects signal state live", async () => { - process.env.DATABRICKS_JOB_ETL = "123"; - - const { Context } = await import("../../../workspace-client"); - const mockContext = Context as unknown as ReturnType; - mockContext.mockClear(); - + test("connector forwards the abort signal as CallOptions", async () => { const { JobsConnector } = await import("../../../connectors/jobs"); const connector = new JobsConnector({}); - jobsApi.get.mockResolvedValue({ job_id: 123 }); + jobsApi.getJob.mockResolvedValue({ jobId: 123n }); const controller = new AbortController(); await connector.getJob( @@ -681,12 +668,10 @@ describe("JobsPlugin", () => { controller.signal, ); - const ctorArg = mockContext.mock.calls.at(-1)?.[0] as { - cancellationToken: { isCancellationRequested: boolean }; - }; - expect(ctorArg.cancellationToken.isCancellationRequested).toBe(false); - controller.abort(); - expect(ctorArg.cancellationToken.isCancellationRequested).toBe(true); + expect(jobsApi.getJob).toHaveBeenCalledWith( + { jobId: 123n }, + { signal: controller.signal }, + ); }); }); @@ -694,10 +679,10 @@ describe("JobsPlugin", () => { test("runAndWait stops polling when signal is aborted", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); jobsApi.getRun.mockResolvedValue({ - run_id: 42, - state: { life_cycle_state: "RUNNING" }, + runId: 42n, + state: { lifeCycleState: "RUNNING" }, }); const plugin = new JobsPlugin({ pollIntervalMs: 10 }); @@ -805,14 +790,14 @@ describe("JobsPlugin", () => { process.env.DATABRICKS_JOB_ETL = "100"; process.env.DATABRICKS_JOB_ML = "200"; - jobsApi.runNow.mockResolvedValue({ run_id: 1 }); + jobsApi.runNow.mockResolvedValue({ runId: 1n }); const plugin = new JobsPlugin({}); const exported = plugin.exports(); await exported("etl").runNow(); expect(jobsApi.runNow).toHaveBeenCalledWith( - expect.objectContaining({ job_id: 100 }), + expect.objectContaining({ jobId: 100n }), expect.anything(), ); @@ -820,7 +805,7 @@ describe("JobsPlugin", () => { await exported("ml").runNow(); expect(jobsApi.runNow).toHaveBeenCalledWith( - expect.objectContaining({ job_id: 200 }), + expect.objectContaining({ jobId: 200n }), expect.anything(), ); }); @@ -1060,7 +1045,7 @@ describe("injectRoutes", () => { test("returns runId on successful non-streaming run", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); const plugin = new JobsPlugin({}); const routeSpy = vi.spyOn(plugin as any, "route"); @@ -1179,10 +1164,10 @@ describe("injectRoutes", () => { process.env.DATABRICKS_JOB_ETL = "123"; const mockRuns = [ - { run_id: 1, state: { life_cycle_state: "TERMINATED" } }, - { run_id: 2, state: { life_cycle_state: "RUNNING" } }, + { runId: 1n, state: { lifeCycleState: "TERMINATED" } }, + { runId: 2n, state: { lifeCycleState: "RUNNING" } }, ]; - jobsApi.listRuns.mockReturnValue( + jobsApi.listRunsIter.mockReturnValue( (async function* () { for (const run of mockRuns) yield run; })(), @@ -1212,15 +1197,19 @@ describe("injectRoutes", () => { await handler(mockReq, mockRes); + // The modular SDK's camelCase/bigint models go out in the legacy wire shape. expect(mockRes.json).toHaveBeenCalledWith({ - runs: mockRuns, + runs: [ + { run_id: 1, state: { life_cycle_state: "TERMINATED" } }, + { run_id: 2, state: { life_cycle_state: "RUNNING" } }, + ], }); }); test("passes limit query param to listRuns", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.listRuns.mockReturnValue((async function* () {})()); + jobsApi.listRunsIter.mockReturnValue((async function* () {})()); const plugin = new JobsPlugin({}); const routeSpy = vi.spyOn(plugin as any, "route"); @@ -1247,7 +1236,7 @@ describe("injectRoutes", () => { await handler(mockReq, mockRes); // Verify the connector was called with limit 5 - expect(jobsApi.listRuns).toHaveBeenCalledWith( + expect(jobsApi.listRunsIter).toHaveBeenCalledWith( expect.objectContaining({ limit: 5 }), expect.anything(), ); @@ -1259,9 +1248,9 @@ describe("injectRoutes", () => { process.env.DATABRICKS_JOB_ETL = "123"; const mockRun = { - run_id: 42, - job_id: 123, - state: { life_cycle_state: "TERMINATED" }, + runId: 42n, + jobId: 123n, + state: { lifeCycleState: "TERMINATED" }, }; jobsApi.getRun.mockResolvedValue(mockRun); @@ -1289,7 +1278,11 @@ describe("injectRoutes", () => { await handler(mockReq, mockRes); - expect(mockRes.json).toHaveBeenCalledWith(mockRun); + expect(mockRes.json).toHaveBeenCalledWith({ + run_id: 42, + job_id: 123, + state: { life_cycle_state: "TERMINATED" }, + }); }); test("returns 400 for invalid runId", async () => { @@ -1330,7 +1323,7 @@ describe("injectRoutes", () => { process.env.DATABRICKS_JOB_ETL = "123"; // Run exists upstream but is owned by job 456, not the configured 123. - jobsApi.getRun.mockResolvedValue({ run_id: 99, job_id: 456 }); + jobsApi.getRun.mockResolvedValue({ runId: 99n, jobId: 456n }); const plugin = new JobsPlugin({}); const routeSpy = vi.spyOn(plugin as any, "route"); @@ -1369,10 +1362,10 @@ describe("injectRoutes", () => { process.env.DATABRICKS_JOB_ETL = "123"; const mockRun = { - run_id: 42, - state: { life_cycle_state: "TERMINATED" }, + runId: 42n, + state: { lifeCycleState: "TERMINATED" }, }; - jobsApi.listRuns.mockReturnValue( + jobsApi.listRunsIter.mockReturnValue( (async function* () { yield mockRun; })(), @@ -1404,14 +1397,14 @@ describe("injectRoutes", () => { expect(mockRes.json).toHaveBeenCalledWith({ status: "TERMINATED", - run: mockRun, + run: { run_id: 42, state: { life_cycle_state: "TERMINATED" } }, }); }); test("returns null status when no runs exist", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.listRuns.mockReturnValue((async function* () {})()); + jobsApi.listRunsIter.mockReturnValue((async function* () {})()); const plugin = new JobsPlugin({}); const routeSpy = vi.spyOn(plugin as any, "route"); @@ -1448,7 +1441,7 @@ describe("injectRoutes", () => { test("cancels run and returns 204", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.getRun.mockResolvedValue({ run_id: 42, job_id: 123 }); + jobsApi.getRun.mockResolvedValue({ runId: 42n, jobId: 123n }); jobsApi.cancelRun.mockResolvedValue(undefined); const plugin = new JobsPlugin({}); @@ -1519,7 +1512,7 @@ describe("injectRoutes", () => { process.env.DATABRICKS_JOB_ETL = "123"; // Pre-flight getRun reports a run owned by a different job. - jobsApi.getRun.mockResolvedValue({ run_id: 99, job_id: 456 }); + jobsApi.getRun.mockResolvedValue({ runId: 99n, jobId: 456n }); jobsApi.cancelRun.mockResolvedValue(undefined); const plugin = new JobsPlugin({}); @@ -1701,7 +1694,7 @@ describe("injectRoutes", () => { test("allows exactly MAX_UNVALIDATED_PARAM_KEYS (50) keys without schema", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); const plugin = new JobsPlugin({ jobs: { etl: { taskType: "notebook" } }, @@ -1745,7 +1738,7 @@ describe("injectRoutes", () => { test("allows undefined params", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.runNow.mockResolvedValue({ run_id: 42 }); + jobsApi.runNow.mockResolvedValue({ runId: 42n }); const plugin = new JobsPlugin({}); const routeSpy = vi.spyOn(plugin as any, "route"); @@ -1826,7 +1819,7 @@ describe("injectRoutes", () => { test("GET /:jobKey/runs returns upstream status on failure", async () => { process.env.DATABRICKS_JOB_ETL = "123"; - jobsApi.listRuns.mockImplementation(() => { + jobsApi.listRunsIter.mockImplementation(() => { throw createApiError({ statusCode: 401, message: "Unauthorized", @@ -1865,7 +1858,7 @@ describe("injectRoutes", () => { process.env.DATABRICKS_JOB_ETL = "123"; // Pre-flight succeeds so we reach the actual cancel call. - jobsApi.getRun.mockResolvedValue({ run_id: 42, job_id: 123 }); + jobsApi.getRun.mockResolvedValue({ runId: 42n, jobId: 123n }); const error = new Error("Forbidden"); (error as any).statusCode = 403; jobsApi.cancelRun.mockRejectedValue(error); diff --git a/packages/shared/package.json b/packages/shared/package.json index 6002e4693..01938030b 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -52,6 +52,7 @@ "@databricks/sdk-core": "0.51.0", "@databricks/sdk-experimental": "0.17.0", "@databricks/sdk-genie": "0.54.0", + "@databricks/sdk-jobs": "0.57.0", "@databricks/sdk-options": "0.51.0", "@databricks/sdk-scim": "0.51.0", "@databricks/sdk-statementexecution": "0.52.0", diff --git a/packages/shared/src/workspace-client/client.ts b/packages/shared/src/workspace-client/client.ts index 1552392e5..92ca56b11 100644 --- a/packages/shared/src/workspace-client/client.ts +++ b/packages/shared/src/workspace-client/client.ts @@ -20,6 +20,8 @@ import { type ScimClient, buildGenieClient, type GenieClient, + buildJobsClient, + type JobsClient, type StatementExecutionClient, type WarehousesClient, type WorkspaceAuth, @@ -35,6 +37,7 @@ export class AppKitWorkspaceClient implements WorkspaceClient { #auth?: WorkspaceAuth; #currentUser?: ScimClient; #genie?: GenieClient; + #jobs?: JobsClient; constructor(opts: WorkspaceClientOptions) { this.#opts = opts; @@ -60,8 +63,12 @@ export class AppKitWorkspaceClient implements WorkspaceClient { return this.#genie; } - get jobs() { - return this.#getLegacy().jobs; + // Migrated to the modular SDK — built lazily, independent of the legacy client. + get jobs(): JobsClient { + if (!this.#jobs) { + this.#jobs = buildJobsClient(this.#opts); + } + return this.#jobs; } // Migrated to the modular SDK — built lazily, independent of the legacy client. diff --git a/packages/shared/src/workspace-client/modular.ts b/packages/shared/src/workspace-client/modular.ts index 4e02de2f8..d5f897c16 100644 --- a/packages/shared/src/workspace-client/modular.ts +++ b/packages/shared/src/workspace-client/modular.ts @@ -8,9 +8,9 @@ * * Migrated services are built here as per-service clients; the facade delegates * their accessors to these instead of the legacy monolithic client. Currently - * `warehouses` and `statementExecution` are migrated, plus the auth + raw-request - * seam ({@link buildWorkspaceAuth}); every other service still routes through - * `legacy.ts`. + * `warehouses`, `statementExecution`, `currentUser` (SCIM), `genie` and `jobs` + * are migrated, plus the auth + raw-request seam ({@link buildWorkspaceAuth}); + * every other service still routes through `legacy.ts`. * * NOTE: statementExecution relies on a pinned pnpm patch * (`patches/@databricks__sdk-statementexecution@0.46.0.patch`) that restores the @@ -46,6 +46,7 @@ import { } from "@databricks/sdk-core/http"; import { type ProfileOptions, resolve } from "@databricks/sdk-core/profiles"; import type { GenieClient } from "@databricks/sdk-genie/v1"; +import type { JobsClient } from "@databricks/sdk-jobs/v2"; import type { ClientOptions } from "@databricks/sdk-options/client"; import { ScimClient } from "@databricks/sdk-scim/v1"; import { StatementExecutionClient } from "@databricks/sdk-statementexecution/v1"; @@ -544,8 +545,17 @@ export function buildGenieClient(opts: WorkspaceClientOptions): GenieClient { ); } +/** Build a modular Jobs (API 2.2) client from wrapper options. */ +export function buildJobsClient(opts: WorkspaceClientOptions): JobsClient { + return lazyClient( + () => import("@databricks/sdk-jobs/v2").then((m) => m.JobsClient), + mapToClientOptions(opts), + ); +} + // ── Client type re-exports (for the facade accessor types) ─────────────── export type { GenieClient } from "@databricks/sdk-genie/v1"; +export type { JobsClient } from "@databricks/sdk-jobs/v2"; export type { ScimClient } from "@databricks/sdk-scim/v1"; export type { StatementExecutionClient } from "@databricks/sdk-statementexecution/v1"; export type { WarehousesClient } from "@databricks/sdk-warehouses/v1"; @@ -574,6 +584,13 @@ export type { GenieGetMessageQueryResultResponse, GenieMessage, } from "@databricks/sdk-genie/v1"; +export type { + GetJobRequest, + GetRunRequest, + ListRunsRequest, + RunNowRequest, + SubmitRunRequest, +} from "@databricks/sdk-jobs/v2"; export type { EndpointHealth, EndpointInfo, diff --git a/packages/shared/src/workspace-client/types.ts b/packages/shared/src/workspace-client/types.ts index 7dd98c30f..40327c840 100644 --- a/packages/shared/src/workspace-client/types.ts +++ b/packages/shared/src/workspace-client/types.ts @@ -20,6 +20,7 @@ import type { WorkspaceAuth, ScimClient, GenieClient, + JobsClient, } from "./modular"; // Legacy SDK type namespaces for un-migrated services, re-exported so AppKit @@ -30,8 +31,11 @@ import type { // as `sql.EndpointInfo[]`. It is not on the modular `listWarehouses` because // that request has no `skip_cannot_use` filter, so it could pick a warehouse // the caller can't use. Statement + warehouse service types come from `./modular`. +// `jobs` stays as the wire-shape (snake_case, `number` IDs) type of the jobs +// plugin's public API; appkit's jobs connector translates the modular client's +// camelCase/`bigint` models back to it. export type { files, jobs, serving, sql } from "@databricks/sdk-experimental"; -// Modular SDK client + model types (warehouses, statementExecution, genie). +// Modular SDK client + model types (warehouses, statementExecution, genie, jobs). export type * from "./modular"; /** @@ -52,8 +56,8 @@ export interface WorkspaceClient extends WorkspaceAuth { /** Genie (modular SDK). */ readonly genie: GenieClient; - /** Jobs. */ - readonly jobs: LegacyWorkspaceClient["jobs"]; + /** Jobs (modular SDK, Jobs API 2.2). */ + readonly jobs: JobsClient; /** Statement Execution (modular SDK). */ readonly statementExecution: StatementExecutionClient; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index fd69b375f..c1b0f795e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -292,6 +292,9 @@ importers: '@databricks/sdk-genie': specifier: 0.54.0 version: 0.54.0(patch_hash=664f3e13b155eb1b5499486a4f6fc1ab9a6a502e2c6b200d06749bd25dd9ae54) + '@databricks/sdk-jobs': + specifier: 0.57.0 + version: 0.57.0 '@databricks/sdk-options': specifier: 0.51.0 version: 0.51.0 @@ -628,6 +631,9 @@ importers: '@databricks/sdk-genie': specifier: 0.54.0 version: 0.54.0(patch_hash=664f3e13b155eb1b5499486a4f6fc1ab9a6a502e2c6b200d06749bd25dd9ae54) + '@databricks/sdk-jobs': + specifier: 0.57.0 + version: 0.57.0 '@databricks/sdk-options': specifier: 0.51.0 version: 0.51.0 @@ -2025,6 +2031,10 @@ packages: resolution: {integrity: sha512-xT+dyuXocgwPJ5VA1Kr5/dSdKzzR9hRAkEOybeO2i+v891kuuFzVd/M9OM31PtO2/YZp95vdFH5Sb+jVGtm4nw==} engines: {node: '>=22.0.0'} + '@databricks/sdk-jobs@0.57.0': + resolution: {integrity: sha512-LUBLNRhJq0OtUUU6+2T/zRzZMEd4UL0lVm7qTxLIfuDPDNDPy/EHGs3rQ1cXt1Fv0t3Rik7QT/SueZixSsH/Bg==} + engines: {node: '>=22.0.0'} + '@databricks/sdk-options@0.51.0': resolution: {integrity: sha512-p5uBh64Y1onnvwlEwfRsmKmGRI23Y4Ly8fZiCgNUwVDe4Z9d0ctI0mo1v4gnDqlmBaBml3TZfqn5BJBROCXsEw==} engines: {node: '>=22.0.0'} @@ -14346,6 +14356,15 @@ snapshots: json-bigint: 1.0.0 zod: 4.3.6 + '@databricks/sdk-jobs@0.57.0': + dependencies: + '@databricks/sdk-auth': 0.51.0 + '@databricks/sdk-core': 0.51.0 + '@databricks/sdk-options': 0.51.0 + '@js-temporal/polyfill': 0.5.1 + json-bigint: 1.0.0 + zod: 4.3.6 + '@databricks/sdk-options@0.51.0': dependencies: '@databricks/sdk-auth': 0.51.0 From 6301dd35a713a92b5f2f9f9f92b93db7e0916af7 Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Wed, 14 Oct 2026 12:50:36 +0200 Subject: [PATCH 2/3] refactor(appkit): expose the modular jobs models directly Drop the snake_case wire-shape translation in the jobs connector and use the modular SDK's camelCase requests and models end to end. Only bigint int64s are converted to number (JSON/SSE/cache can't serialize bigint). - Connector types: JobRun, JobBaseRun, JobRunOutput, JobDetails (Numeric over the SDK models); runNow returns { runId }. - mapParams emits notebookParams/pythonNamedParams/... . - Playground, template and docs updated to the camelCase fields. - Drop the unused legacy `jobs` type namespace re-export. Co-authored-by: Isaac Signed-off-by: MarioCadenas --- .../client/src/routes/jobs.route.tsx | 30 +-- docs/docs/plugins/jobs.md | 12 +- packages/appkit/src/connectors/jobs/client.ts | 191 +++++------------- packages/appkit/src/connectors/jobs/index.ts | 8 +- .../src/connectors/jobs/tests/client.test.ts | 51 ++--- packages/appkit/src/connectors/jobs/types.ts | 24 +++ packages/appkit/src/plugins/jobs/params.ts | 18 +- packages/appkit/src/plugins/jobs/plugin.ts | 63 +++--- .../src/plugins/jobs/tests/plugin.test.ts | 40 ++-- packages/appkit/src/plugins/jobs/types.ts | 23 ++- packages/appkit/src/workspace-client/index.ts | 2 +- packages/shared/src/workspace-client/index.ts | 2 +- .../shared/src/workspace-client/modular.ts | 6 + packages/shared/src/workspace-client/types.ts | 7 +- template/client/src/pages/jobs/JobsPage.tsx | 30 +-- 15 files changed, 217 insertions(+), 290 deletions(-) diff --git a/apps/dev-playground/client/src/routes/jobs.route.tsx b/apps/dev-playground/client/src/routes/jobs.route.tsx index efcb945b1..b57b4931c 100644 --- a/apps/dev-playground/client/src/routes/jobs.route.tsx +++ b/apps/dev-playground/client/src/routes/jobs.route.tsx @@ -9,10 +9,10 @@ export const Route = createFileRoute("/jobs")({ const MAX_STREAM_LOG = 500; interface Run { - run_id: number; - state?: { life_cycle_state?: string; result_state?: string }; - start_time?: number; - end_time?: number; + runId: number; + state?: { lifeCycleState?: string; resultState?: string }; + startTime?: number; + endTime?: number; } function formatTime(ts?: number) { @@ -211,8 +211,8 @@ function JobsRoute() {

{status.run && (

- Run #{status.run.run_id} — started{" "} - {formatTime(status.run.start_time)} + Run #{status.run.runId} — started{" "} + {formatTime(status.run.startTime)}

)} @@ -235,27 +235,27 @@ function JobsRoute() { {runs.map((run) => ( - - {run.run_id} + + {run.runId} - {run.state?.life_cycle_state ?? "—"} + {run.state?.lifeCycleState ?? "—"} - {run.state?.result_state ?? "—"} + {run.state?.resultState ?? "—"} - {formatTime(run.start_time)} + {formatTime(run.startTime)} - {run.state?.life_cycle_state === "RUNNING" && ( + {run.state?.lifeCycleState === "RUNNING" && ( diff --git a/docs/docs/plugins/jobs.md b/docs/docs/plugins/jobs.md index d736dcfb7..97d97c5ab 100644 --- a/docs/docs/plugins/jobs.md +++ b/docs/docs/plugins/jobs.md @@ -97,11 +97,11 @@ When `taskType` is set, the plugin maps validated parameters to the correct SDK | Task type | SDK field | Parameter shape | |-----------|-----------|-----------------| -| `notebook` | `notebook_params` | `Record` — values coerced to string | -| `python_wheel` | `python_named_params` | `Record` — values coerced to string | -| `python_script` | `python_params` | `{ args: string[] }` — positional args | -| `spark_jar` | `jar_params` | `{ args: string[] }` — positional args | -| `sql` | `sql_params` | `Record` — values coerced to string | +| `notebook` | `notebookParams` | `Record` — values coerced to string | +| `python_wheel` | `pythonNamedParams` | `Record` — values coerced to string | +| `python_script` | `pythonParams` | `{ args: string[] }` — positional args | +| `spark_jar` | `jarParams` | `{ args: string[] }` — positional args | +| `sql` | `sqlParams` | `Record` — values coerced to string | | `dbt` | — | No parameters accepted | ```ts @@ -198,7 +198,7 @@ const etl = AppKit.jobs("etl"); // Trigger a run const result = await etl.runNow({ startDate: "2025-01-01" }); if (result.ok) { - console.log("Run ID:", result.data.run_id); + console.log("Run ID:", result.data.runId); } // Trigger and poll until completion diff --git a/packages/appkit/src/connectors/jobs/client.ts b/packages/appkit/src/connectors/jobs/client.ts index ab048a202..3b3aea1ac 100644 --- a/packages/appkit/src/connectors/jobs/client.ts +++ b/packages/appkit/src/connectors/jobs/client.ts @@ -9,93 +9,40 @@ import { SpanStatusCode, TelemetryManager, } from "../../telemetry"; +import { isPlainObject } from "../../utils/is-plain-object"; import type { + CancelRunRequest, GetJobRequest, + GetRunOutputRequest, GetRunRequest, - jobs, ListRunsRequest, RunNowRequest, SubmitRunRequest, WorkspaceClient, } from "../../workspace-client"; -import type { JobsConnectorConfig } from "./types"; +import type { + JobBaseRun, + JobDetails, + JobRun, + JobRunOutput, + JobsConnectorConfig, + Numeric, +} from "./types"; const logger = createLogger("connectors:jobs"); /** - * `Record` fields of the Jobs model. Their keys are user data - * (notebook params, tags, Spark conf), so they are copied verbatim instead of - * re-cased. - */ -const MAP_FIELDS = new Set([ - "artifactsHeaders", - "baseParameters", - "customTags", - "filters", - "jobParameters", - "namedParameters", - "notebookBaseParameters", - "notebookParams", - "parameters", - "pipelineTaskParameters", - "pythonNamedParams", - "sparkConf", - "sparkEnvVars", - "sqlParams", - "tags", - "variables", - "violations", -]); - -/** The one model field whose camelCase name doesn't round-trip to its wire key. */ -const WIRE_KEY_OVERRIDES: Record = { - pipelineTaskParameters: "parameters", -}; - -function isPlainObject(value: unknown): value is Record { - return value !== null && typeof value === "object" && !Array.isArray(value); -} - -/** - * Modular model → the legacy wire shape the plugin's public API (HTTP JSON, SSE, - * cache) has always exposed: snake_case keys and `number` int64s. `bigint` would - * make `JSON.stringify` throw. `Number()` loses precision past 2^53, exactly as - * the legacy SDK's plain `JSON.parse` did. + * The modular SDK types int64s as `bigint`, which `JSON.stringify` rejects, and + * every jobs response ends up in HTTP JSON, SSE or the cache. `Number()` loses + * precision past 2^53, as the legacy SDK's plain `JSON.parse` did. */ -function toWire(value: unknown, verbatimKeys = false): unknown { - if (typeof value === "bigint") return Number(value); - if (Array.isArray(value)) return value.map((v) => toWire(v)); - if (!isPlainObject(value)) return value; +function toNumbers(value: T): Numeric { + if (typeof value === "bigint") return Number(value) as Numeric; + if (Array.isArray(value)) return value.map(toNumbers) as Numeric; + if (!isPlainObject(value)) return value as Numeric; return Object.fromEntries( - Object.entries(value).map(([key, v]) => - verbatimKeys - ? [key, v] - : [ - WIRE_KEY_OVERRIDES[key] ?? - key.replace(/[A-Z]/g, (c) => `_${c.toLowerCase()}`), - toWire(v, MAP_FIELDS.has(key)), - ], - ), - ); -} - -/** - * Legacy snake_case request → modular camelCase request. int64 fields (`job_id`, - * `run_id`) still need an explicit `BigInt()` at the call site: the SDK's - * marshal schemas reject a `number` there. - */ -function fromWire(value: unknown, verbatimKeys = false): unknown { - if (Array.isArray(value)) return value.map((v) => fromWire(v)); - if (!isPlainObject(value)) return value; - return Object.fromEntries( - Object.entries(value).map(([key, v]) => { - if (verbatimKeys) return [key, v]; - const camel = key.replace(/_([a-z0-9])/g, (_, c: string) => - c.toUpperCase(), - ); - return [camel, fromWire(v, MAP_FIELDS.has(camel))]; - }), - ); + Object.entries(value).map(([k, v]) => [k, toNumbers(v)]), + ) as Numeric; } export class JobsConnector { @@ -131,100 +78,69 @@ export class JobsConnector { async submitRun( workspaceClient: WorkspaceClient, - request: jobs.SubmitRun, + request: SubmitRunRequest, signal?: AbortSignal, - ): Promise { + ): Promise<{ runId: number }> { return this._callApi("submit", async () => { - const waiter = await workspaceClient.jobs.submitRun( - fromWire(request) as SubmitRunRequest, - { signal }, - ); - return { run_id: Number(waiter.runId) }; + const waiter = await workspaceClient.jobs.submitRun(request, { signal }); + return { runId: Number(waiter.runId) }; }); } async runNow( workspaceClient: WorkspaceClient, - request: jobs.RunNow, + request: RunNowRequest, signal?: AbortSignal, - ): Promise { + ): Promise<{ runId: number }> { return this._callApi("runNow", async () => { - const waiter = await workspaceClient.jobs.runNow( - { - ...(fromWire(request) as RunNowRequest), - jobId: BigInt(request.job_id), - }, - { signal }, - ); - // The waiter only exposes runId; the API documents number_in_job as - // "set to the same value as run_id". - const runId = Number(waiter.runId); - return { run_id: runId, number_in_job: runId }; + const waiter = await workspaceClient.jobs.runNow(request, { signal }); + return { runId: Number(waiter.runId) }; }); } async getRun( workspaceClient: WorkspaceClient, - request: jobs.GetRunRequest, + request: GetRunRequest, signal?: AbortSignal, - ): Promise { - return this._callApi("getRun", async () => { - const run = await workspaceClient.jobs.getRun( - { - ...(fromWire(request) as GetRunRequest), - runId: BigInt(request.run_id), - }, - { signal }, - ); - return toWire(run) as jobs.Run; - }); + ): Promise { + return this._callApi("getRun", async () => + toNumbers(await workspaceClient.jobs.getRun(request, { signal })), + ); } async getRunOutput( workspaceClient: WorkspaceClient, - request: jobs.GetRunOutputRequest, + request: GetRunOutputRequest, signal?: AbortSignal, - ): Promise { - return this._callApi("getRunOutput", async () => { - const output = await workspaceClient.jobs.getRunOutput( - { runId: BigInt(request.run_id) }, - { signal }, - ); - return toWire(output) as jobs.RunOutput; - }); + ): Promise { + return this._callApi("getRunOutput", async () => + toNumbers(await workspaceClient.jobs.getRunOutput(request, { signal })), + ); } async cancelRun( workspaceClient: WorkspaceClient, - request: jobs.CancelRun, + request: CancelRunRequest, signal?: AbortSignal, ): Promise { await this._callApi("cancelRun", async () => { - await workspaceClient.jobs.cancelRun( - { runId: BigInt(request.run_id) }, - { signal }, - ); + await workspaceClient.jobs.cancelRun(request, { signal }); }); } async listRuns( workspaceClient: WorkspaceClient, - request: jobs.ListRunsRequest, + request: ListRunsRequest, signal?: AbortSignal, - ): Promise { + ): Promise { return this._callApi("listRuns", async () => { - const runs: jobs.BaseRun[] = []; + const runs: JobBaseRun[] = []; const limit = Math.max(1, Math.min(request.limit ?? 100, 100)); for await (const run of workspaceClient.jobs.listRunsIter( - { - ...(fromWire(request) as ListRunsRequest), - jobId: - request.job_id === undefined ? undefined : BigInt(request.job_id), - limit, - }, + { ...request, limit }, { signal }, )) { - runs.push(toWire(run) as jobs.BaseRun); + runs.push(toNumbers(run)); if (runs.length >= limit) break; } return runs; @@ -233,19 +149,12 @@ export class JobsConnector { async getJob( workspaceClient: WorkspaceClient, - request: jobs.GetJobRequest, + request: GetJobRequest, signal?: AbortSignal, - ): Promise { - return this._callApi("getJob", async () => { - const job = await workspaceClient.jobs.getJob( - { - ...(fromWire(request) as GetJobRequest), - jobId: BigInt(request.job_id), - }, - { signal }, - ); - return toWire(job) as jobs.Job; - }); + ): Promise { + return this._callApi("getJob", async () => + toNumbers(await workspaceClient.jobs.getJob(request, { signal })), + ); } private async _callApi( diff --git a/packages/appkit/src/connectors/jobs/index.ts b/packages/appkit/src/connectors/jobs/index.ts index efb4753a8..49359ab2c 100644 --- a/packages/appkit/src/connectors/jobs/index.ts +++ b/packages/appkit/src/connectors/jobs/index.ts @@ -1,2 +1,8 @@ export { JobsConnector } from "./client"; -export type { JobsConnectorConfig } from "./types"; +export type { + JobBaseRun, + JobDetails, + JobRun, + JobRunOutput, + JobsConnectorConfig, +} from "./types"; diff --git a/packages/appkit/src/connectors/jobs/tests/client.test.ts b/packages/appkit/src/connectors/jobs/tests/client.test.ts index 542175284..4ade28d39 100644 --- a/packages/appkit/src/connectors/jobs/tests/client.test.ts +++ b/packages/appkit/src/connectors/jobs/tests/client.test.ts @@ -2,59 +2,38 @@ import { describe, expect, test, vi } from "vitest"; import { JobsConnector } from "../client"; -// The modular Jobs SDK returns camelCase models with `bigint` int64s, which -// `JSON.stringify` rejects. The connector must hand the plugin the legacy wire -// shape (snake_case, `number`) its public HTTP/SSE/cache surface has always had. -describe("JobsConnector wire-shape translation", () => { - test("getRun returns a JSON-serializable snake_case run with numeric IDs", async () => { +// The modular Jobs SDK types int64s as `bigint`, which `JSON.stringify` rejects. +// Every connector response ends up in HTTP JSON, SSE or the cache. +describe("JobsConnector", () => { + test("getRun returns a JSON-serializable run with numeric int64s", async () => { const getRun = vi.fn().mockResolvedValue({ runId: 9007199254740991n, jobId: 123n, startTime: 1700000000000n, state: { lifeCycleState: "TERMINATED" }, - overridingParameters: { notebookParams: { myParam: "x" } }, - tasks: [ - { - taskKey: "t1", - pipelineTask: { pipelineTaskParameters: { fullRefresh: "true" } }, - }, - ], + tasks: [{ taskKey: "t1", attemptNumber: 0n }], }); const client = { jobs: { getRun } } as never; - const run = await new JobsConnector({}).getRun(client, { run_id: 42 }); + const run = await new JobsConnector({}).getRun(client, { runId: 42n }); expect(getRun).toHaveBeenCalledWith({ runId: 42n }, { signal: undefined }); expect(JSON.parse(JSON.stringify(run))).toEqual({ - run_id: 9007199254740991, - job_id: 123, - start_time: 1700000000000, - state: { life_cycle_state: "TERMINATED" }, - // Map keys are user data: copied verbatim, not re-cased. - overriding_parameters: { notebook_params: { myParam: "x" } }, - tasks: [ - { - task_key: "t1", - pipeline_task: { parameters: { fullRefresh: "true" } }, - }, - ], + runId: 9007199254740991, + jobId: 123, + startTime: 1700000000000, + state: { lifeCycleState: "TERMINATED" }, + tasks: [{ taskKey: "t1", attemptNumber: 0 }], }); }); - test("runNow camelCases the request without touching param keys", async () => { + test("runNow returns the waiter's runId as a number", async () => { const runNow = vi.fn().mockResolvedValue({ runId: 7n }); const client = { jobs: { runNow } } as never; - const result = await new JobsConnector({}).runNow(client, { - job_id: 123, - notebook_params: { my_param: "v" }, - }); + const result = await new JobsConnector({}).runNow(client, { jobId: 123n }); - expect(runNow).toHaveBeenCalledWith( - { jobId: 123n, notebookParams: { my_param: "v" } }, - { signal: undefined }, - ); - expect(result).toEqual({ run_id: 7, number_in_job: 7 }); + expect(result).toEqual({ runId: 7 }); }); test("surfaces the modular ApiError's httpStatusCode as statusCode", async () => { @@ -68,7 +47,7 @@ describe("JobsConnector wire-shape translation", () => { const client = { jobs: { getRun } } as never; await expect( - new JobsConnector({}).getRun(client, { run_id: 1 }), + new JobsConnector({}).getRun(client, { runId: 1n }), ).rejects.toMatchObject({ statusCode: 404, message: "not found" }); }); }); diff --git a/packages/appkit/src/connectors/jobs/types.ts b/packages/appkit/src/connectors/jobs/types.ts index dd7dfbbe5..01d255a92 100644 --- a/packages/appkit/src/connectors/jobs/types.ts +++ b/packages/appkit/src/connectors/jobs/types.ts @@ -1,5 +1,29 @@ import type { TelemetryOptions } from "shared"; +import type { + BaseRun, + GetJobResponse, + GetRunOutputResponse, + GetRunResponse, +} from "../../workspace-client"; + export interface JobsConnectorConfig { telemetry?: TelemetryOptions; } + +/** A modular SDK model with its `bigint` int64s as `number` (JSON-safe). */ +export type Numeric = T extends bigint + ? number + : // Before the object branch: SDK enums are `"X" | (string & {})`. + T extends string | number | boolean | null | undefined + ? T + : T extends readonly (infer U)[] + ? Numeric[] + : T extends object + ? { [K in keyof T]: Numeric } + : T; + +export type JobRun = Numeric; +export type JobBaseRun = Numeric; +export type JobRunOutput = Numeric; +export type JobDetails = Numeric; diff --git a/packages/appkit/src/plugins/jobs/params.ts b/packages/appkit/src/plugins/jobs/params.ts index 757107446..75b7545bb 100644 --- a/packages/appkit/src/plugins/jobs/params.ts +++ b/packages/appkit/src/plugins/jobs/params.ts @@ -21,36 +21,34 @@ export function mapParams( ): Record { switch (taskType) { case "notebook": - // notebook_params expects Record, values coerced to string + // notebookParams expects Record, values coerced to string assertPrimitiveValues(params); return { - notebook_params: Object.fromEntries( + notebookParams: Object.fromEntries( Object.entries(params).map(([k, v]) => [k, String(v)]), ), }; case "python_wheel": assertPrimitiveValues(params); return { - python_named_params: Object.fromEntries( + pythonNamedParams: Object.fromEntries( Object.entries(params).map(([k, v]) => [k, String(v)]), ), }; case "python_script": - // python_params expects string[] (positional args) + // pythonParams expects string[] (positional args) return { - python_params: Array.isArray(params.args) - ? params.args.map(String) - : [], + pythonParams: Array.isArray(params.args) ? params.args.map(String) : [], }; case "spark_jar": - // jar_params expects string[] + // jarParams expects string[] return { - jar_params: Array.isArray(params.args) ? params.args.map(String) : [], + jarParams: Array.isArray(params.args) ? params.args.map(String) : [], }; case "sql": assertPrimitiveValues(params); return { - sql_params: Object.fromEntries( + sqlParams: Object.fromEntries( Object.entries(params).map(([k, v]) => [k, String(v)]), ), }; diff --git a/packages/appkit/src/plugins/jobs/plugin.ts b/packages/appkit/src/plugins/jobs/plugin.ts index d9f4913b1..9bc905929 100644 --- a/packages/appkit/src/plugins/jobs/plugin.ts +++ b/packages/appkit/src/plugins/jobs/plugin.ts @@ -8,7 +8,13 @@ import type { } from "shared"; import { toJSONSchema } from "zod"; -import { JobsConnector } from "../../connectors/jobs"; +import { + type JobBaseRun, + type JobDetails, + type JobRun, + type JobRunOutput, + JobsConnector, +} from "../../connectors/jobs"; import { getCurrentPrincipalId, getWorkspaceClient } from "../../context"; import { ExecutionError, ValidationError } from "../../errors"; import { createLogger } from "../../logging/logger"; @@ -16,7 +22,6 @@ import type { ExecutionResult } from "../../plugin"; import { Plugin, toPlugin } from "../../plugin"; import type { ResourceRequirement } from "../../registry"; import { defineManifest, ResourceType } from "../../registry"; -import type { jobs as jobsTypes } from "../../workspace-client"; import { JOBS_READ_DEFAULTS, JOBS_STREAM_DEFAULTS, @@ -270,26 +275,26 @@ class JobsPlugin extends Plugin { ): Promise | null> => { const result = await self.execute( async (signal) => - self.connector.getRun(client, { run_id: runId }, signal), + self.connector.getRun(client, { runId: BigInt(runId) }, signal), self._readSettings(["jobs:getRun", jobKey, runId]), userKey, ); if (!result.ok) return errorResult(result.status); - if (result.data.job_id !== jobId) return errorResult(404); + if (result.data.jobId !== jobId) return errorResult(404); return null; }; return { runNow: async ( params?: Record, - ): Promise> => { + ): Promise> => { const sdkFields = self._validateAndMap(jobKey, params); const result = await self.execute( async (signal) => self.connector.runNow( client, - { ...sdkFields, job_id: jobId }, + { ...sdkFields, jobId: BigInt(jobId) }, signal, ), self._writeSettings(), @@ -308,7 +313,7 @@ class JobsPlugin extends Plugin { async (signal) => self.connector.runNow( client, - { ...sdkFields, job_id: jobId }, + { ...sdkFields, jobId: BigInt(jobId) }, signal, ), self._writeSettings(), @@ -318,9 +323,9 @@ class JobsPlugin extends Plugin { if (!runResult.ok) { throw new ExecutionError("Failed to trigger job run"); } - const runId = runResult.data.run_id; + const runId = runResult.data.runId; if (!runId) { - throw new Error("runNow did not return a run_id"); + throw new Error("runNow did not return a runId"); } const basePollInterval = @@ -339,7 +344,7 @@ class JobsPlugin extends Plugin { const runStatusResult = await self.execute( async (signal) => - self.connector.getRun(client, { run_id: runId }, signal), + self.connector.getRun(client, { runId: BigInt(runId) }, signal), { default: { ...JOBS_READ_DEFAULTS, @@ -354,7 +359,7 @@ class JobsPlugin extends Plugin { ); } const run = runStatusResult.data; - const state = run.state?.life_cycle_state; + const state = run.state?.lifeCycleState; yield { status: state, timestamp: Date.now(), run }; @@ -369,14 +374,12 @@ class JobsPlugin extends Plugin { } }, - lastRun: async (): Promise< - ExecutionResult - > => { + lastRun: async (): Promise> => { const result = await self.execute( async (signal) => self.connector.listRuns( client, - { job_id: jobId, limit: 1 }, + { jobId: BigInt(jobId), limit: 1 }, signal, ), self._readSettings(["jobs:lastRun", jobKey]), @@ -388,12 +391,12 @@ class JobsPlugin extends Plugin { listRuns: async (options?: { limit?: number; - }): Promise> => { + }): Promise> => { const result = await self.execute( async (signal) => self.connector.listRuns( client, - { job_id: jobId, limit: options?.limit }, + { jobId: BigInt(jobId), limit: options?.limit }, signal, ), self._readSettings([ @@ -406,28 +409,30 @@ class JobsPlugin extends Plugin { return result.ok ? result : errorResult(result.status); }, - getRun: async ( - runId: number, - ): Promise> => { + getRun: async (runId: number): Promise> => { const result = await self.execute( async (signal) => - self.connector.getRun(client, { run_id: runId }, signal), + self.connector.getRun(client, { runId: BigInt(runId) }, signal), self._readSettings(["jobs:getRun", jobKey, runId]), userKey, ); if (!result.ok) return errorResult(result.status); - if (result.data.job_id !== jobId) return errorResult(404); + if (result.data.jobId !== jobId) return errorResult(404); return result; }, getRunOutput: async ( runId: number, - ): Promise> => { + ): Promise> => { const scopeError = await verifyRunScope(runId); if (scopeError) return scopeError; const result = await self.execute( async (signal) => - self.connector.getRunOutput(client, { run_id: runId }, signal), + self.connector.getRunOutput( + client, + { runId: BigInt(runId) }, + signal, + ), self._readSettings(["jobs:getRunOutput", jobKey, runId]), userKey, ); @@ -439,17 +444,17 @@ class JobsPlugin extends Plugin { if (scopeError) return scopeError; const result = await self.execute( async (signal) => - self.connector.cancelRun(client, { run_id: runId }, signal), + self.connector.cancelRun(client, { runId: BigInt(runId) }, signal), self._writeSettings(), userKey, ); return result.ok ? result : errorResult(result.status); }, - getJob: async (): Promise> => { + getJob: async (): Promise> => { const result = await self.execute( async (signal) => - self.connector.getJob(client, { job_id: jobId }, signal), + self.connector.getJob(client, { jobId: BigInt(jobId) }, signal), self._readSettings(["jobs:getJob", jobKey]), userKey, ); @@ -567,7 +572,7 @@ class JobsPlugin extends Plugin { this._sendStatusError(res, result.status); return; } - res.json({ runId: result.data.run_id }); + res.json({ runId: result.data.runId }); } } catch (error) { if (error instanceof ValidationError) { @@ -674,7 +679,7 @@ class JobsPlugin extends Plugin { return; } res.json({ - status: result.data?.state?.life_cycle_state ?? null, + status: result.data?.state?.lifeCycleState ?? null, run: result.data ?? null, }); } catch (error) { diff --git a/packages/appkit/src/plugins/jobs/tests/plugin.test.ts b/packages/appkit/src/plugins/jobs/tests/plugin.test.ts index a9fc1a33d..4eb25d1b0 100644 --- a/packages/appkit/src/plugins/jobs/tests/plugin.test.ts +++ b/packages/appkit/src/plugins/jobs/tests/plugin.test.ts @@ -281,7 +281,7 @@ describe("JobsPlugin", () => { const handle = exported("etl"); await handle.runNow({ - notebook_params: { key: "value" }, + notebookParams: { key: "value" }, }); expect(jobsApi.runNow).toHaveBeenCalledWith( @@ -478,9 +478,7 @@ describe("JobsPlugin", () => { const handle = plugin.exports()("etl"); const gen = handle.runAndWait(); - await expect(gen.next()).rejects.toThrow( - "runNow did not return a run_id", - ); + await expect(gen.next()).rejects.toThrow("runNow did not return a runId"); }); }); @@ -588,7 +586,7 @@ describe("JobsPlugin", () => { const result = await handle.runNow(); expect(result.ok).toBe(true); if (result.ok) { - expect(result.data.run_id).toBe(42); + expect(result.data.runId).toBe(42); } }); }); @@ -652,7 +650,7 @@ describe("JobsPlugin", () => { const result = await handle.getRun(42); expect(result.ok).toBe(true); - if (result.ok) expect(result.data.run_id).toBe(42); + if (result.ok) expect(result.data.runId).toBe(42); }); test("connector forwards the abort signal as CallOptions", async () => { @@ -664,7 +662,7 @@ describe("JobsPlugin", () => { const controller = new AbortController(); await connector.getJob( mockClient as never, - { job_id: 123 }, + { jobId: 123n }, controller.signal, ); @@ -837,27 +835,27 @@ describe("defaults", () => { describe("mapParams", () => { test("notebook maps to notebook_params with string coercion", () => { const result = mapParams("notebook", { key: "value", num: 42 }); - expect(result).toEqual({ notebook_params: { key: "value", num: "42" } }); + expect(result).toEqual({ notebookParams: { key: "value", num: "42" } }); }); test("python_wheel maps to python_named_params", () => { const result = mapParams("python_wheel", { arg1: "a", arg2: "b" }); - expect(result).toEqual({ python_named_params: { arg1: "a", arg2: "b" } }); + expect(result).toEqual({ pythonNamedParams: { arg1: "a", arg2: "b" } }); }); test("python_script maps to python_params array", () => { const result = mapParams("python_script", { args: ["a", "b", "c"] }); - expect(result).toEqual({ python_params: ["a", "b", "c"] }); + expect(result).toEqual({ pythonParams: ["a", "b", "c"] }); }); test("spark_jar maps to jar_params array", () => { const result = mapParams("spark_jar", { args: ["x", "y"] }); - expect(result).toEqual({ jar_params: ["x", "y"] }); + expect(result).toEqual({ jarParams: ["x", "y"] }); }); test("sql maps to sql_params Record", () => { const result = mapParams("sql", { p1: "v1", p2: 42 }); - expect(result).toEqual({ sql_params: { p1: "v1", p2: "42" } }); + expect(result).toEqual({ sqlParams: { p1: "v1", p2: "42" } }); }); test("dbt with empty params returns empty object", () => { @@ -878,7 +876,7 @@ describe("mapParams", () => { nil: "null", }); expect(result).toEqual({ - notebook_params: { bool: "true", num: "123", nil: "null" }, + notebookParams: { bool: "true", num: "123", nil: "null" }, }); }); }); @@ -1197,11 +1195,11 @@ describe("injectRoutes", () => { await handler(mockReq, mockRes); - // The modular SDK's camelCase/bigint models go out in the legacy wire shape. + // bigint int64s go out as JSON numbers. expect(mockRes.json).toHaveBeenCalledWith({ runs: [ - { run_id: 1, state: { life_cycle_state: "TERMINATED" } }, - { run_id: 2, state: { life_cycle_state: "RUNNING" } }, + { runId: 1, state: { lifeCycleState: "TERMINATED" } }, + { runId: 2, state: { lifeCycleState: "RUNNING" } }, ], }); }); @@ -1279,9 +1277,9 @@ describe("injectRoutes", () => { await handler(mockReq, mockRes); expect(mockRes.json).toHaveBeenCalledWith({ - run_id: 42, - job_id: 123, - state: { life_cycle_state: "TERMINATED" }, + runId: 42, + jobId: 123, + state: { lifeCycleState: "TERMINATED" }, }); }); @@ -1352,7 +1350,7 @@ describe("injectRoutes", () => { expect(mockRes.status).toHaveBeenCalledWith(404); // Response body must not leak the run object from the other job. expect(mockRes.json).not.toHaveBeenCalledWith( - expect.objectContaining({ job_id: 456 }), + expect.objectContaining({ jobId: 456 }), ); }); }); @@ -1397,7 +1395,7 @@ describe("injectRoutes", () => { expect(mockRes.json).toHaveBeenCalledWith({ status: "TERMINATED", - run: { run_id: 42, state: { life_cycle_state: "TERMINATED" } }, + run: { runId: 42, state: { lifeCycleState: "TERMINATED" } }, }); }); diff --git a/packages/appkit/src/plugins/jobs/types.ts b/packages/appkit/src/plugins/jobs/types.ts index af65f80a8..cfdae6b18 100644 --- a/packages/appkit/src/plugins/jobs/types.ts +++ b/packages/appkit/src/plugins/jobs/types.ts @@ -1,8 +1,13 @@ import type { BasePluginConfig } from "shared"; import type { z } from "zod"; +import type { + JobBaseRun, + JobDetails, + JobRun, + JobRunOutput, +} from "../../connectors/jobs"; import type { ExecutionResult } from "../../plugin"; -import type { jobs } from "../../workspace-client"; /** Supported task types for job parameter mapping. */ export type TaskType = @@ -27,7 +32,7 @@ export interface JobConfig { export interface JobRunStatus { status: string | undefined; timestamp: number; - run: jobs.Run; + run: JobRun; } /** User-facing API for a single configured job. */ @@ -35,26 +40,26 @@ export interface JobAPI { /** Trigger the configured job with validated params. Returns the run response. */ runNow( params?: Record, - ): Promise>; + ): Promise>; /** Trigger and poll until completion, yielding status updates. */ runAndWait( params?: Record, signal?: AbortSignal, ): AsyncGenerator; /** Get the most recent run for this job. */ - lastRun(): Promise>; + lastRun(): Promise>; /** List runs for this job. */ listRuns(options?: { limit?: number; - }): Promise>; + }): Promise>; /** Get a specific run by ID. */ - getRun(runId: number): Promise>; + getRun(runId: number): Promise>; /** Get output of a specific run. */ - getRunOutput(runId: number): Promise>; + getRunOutput(runId: number): Promise>; /** Cancel a specific run. */ cancelRun(runId: number): Promise>; /** Get the job definition. */ - getJob(): Promise>; + getJob(): Promise>; } /** Configuration for the Jobs plugin. */ @@ -74,7 +79,7 @@ export interface IJobsConfig extends BasePluginConfig { * @example * ```ts * // Trigger a configured job - * const { run_id } = await appkit.jobs("etl").runNow(); + * const { runId } = await appkit.jobs("etl").runNow(); * * // Trigger and poll until completion * for await (const status of appkit.jobs("etl").runAndWait()) { diff --git a/packages/appkit/src/workspace-client/index.ts b/packages/appkit/src/workspace-client/index.ts index 39e400c6a..d87a83341 100644 --- a/packages/appkit/src/workspace-client/index.ts +++ b/packages/appkit/src/workspace-client/index.ts @@ -6,7 +6,7 @@ */ export { ApiError, ConfigError, Context, createWorkspaceClient } from "shared"; -// Forwards every wrapper type — legacy service namespaces (files/jobs/serving), +// Forwards every wrapper type — legacy service namespaces (files/serving), // the client option/waiter types, and the modular SDK client + model types // (warehouses, statementExecution). `sql` is gone: its statement + warehouse // types now come from the modular SDK. diff --git a/packages/shared/src/workspace-client/index.ts b/packages/shared/src/workspace-client/index.ts index f037e8ef4..d55adf9db 100644 --- a/packages/shared/src/workspace-client/index.ts +++ b/packages/shared/src/workspace-client/index.ts @@ -21,6 +21,6 @@ export { Time, TimeUnits, } from "./legacy"; -export type { files, jobs, serving, sql, WorkspaceClient } from "./types"; +export type { files, serving, sql, WorkspaceClient } from "./types"; // Modular SDK client + model types (warehouses). export type * from "./modular"; diff --git a/packages/shared/src/workspace-client/modular.ts b/packages/shared/src/workspace-client/modular.ts index d5f897c16..73553c24d 100644 --- a/packages/shared/src/workspace-client/modular.ts +++ b/packages/shared/src/workspace-client/modular.ts @@ -585,8 +585,14 @@ export type { GenieMessage, } from "@databricks/sdk-genie/v1"; export type { + BaseRun, + CancelRunRequest, GetJobRequest, + GetJobResponse, + GetRunOutputRequest, + GetRunOutputResponse, GetRunRequest, + GetRunResponse, ListRunsRequest, RunNowRequest, SubmitRunRequest, diff --git a/packages/shared/src/workspace-client/types.ts b/packages/shared/src/workspace-client/types.ts index 40327c840..f347dbc01 100644 --- a/packages/shared/src/workspace-client/types.ts +++ b/packages/shared/src/workspace-client/types.ts @@ -8,7 +8,7 @@ * service means changing its accessor's type here, its getter in `client.ts`, * and the connector that consumes it. * - * Type-namespace re-exports (`files`, `jobs`, `serving`, `sql`) point at + * Type-namespace re-exports (`files`, `serving`, `sql`) point at * `@databricks/sdk-experimental` so existing AppKit call-site shapes stay * stable. They move to the modular `@databricks/sdk-` model exports * as each service migrates. @@ -31,10 +31,7 @@ import type { // as `sql.EndpointInfo[]`. It is not on the modular `listWarehouses` because // that request has no `skip_cannot_use` filter, so it could pick a warehouse // the caller can't use. Statement + warehouse service types come from `./modular`. -// `jobs` stays as the wire-shape (snake_case, `number` IDs) type of the jobs -// plugin's public API; appkit's jobs connector translates the modular client's -// camelCase/`bigint` models back to it. -export type { files, jobs, serving, sql } from "@databricks/sdk-experimental"; +export type { files, serving, sql } from "@databricks/sdk-experimental"; // Modular SDK client + model types (warehouses, statementExecution, genie, jobs). export type * from "./modular"; diff --git a/template/client/src/pages/jobs/JobsPage.tsx b/template/client/src/pages/jobs/JobsPage.tsx index 7705bedab..32b5b4b1c 100644 --- a/template/client/src/pages/jobs/JobsPage.tsx +++ b/template/client/src/pages/jobs/JobsPage.tsx @@ -11,10 +11,10 @@ import { useCallback, useId, useRef, useState } from 'react'; const MAX_STREAM_LOG = 500; interface Run { - run_id: number; - state?: { life_cycle_state?: string; result_state?: string }; - start_time?: number; - end_time?: number; + runId: number; + state?: { lifeCycleState?: string; resultState?: string }; + startTime?: number; + endTime?: number; } function formatTime(ts?: number) { @@ -205,8 +205,8 @@ export function JobsPage() {

{status.run && (

- Run #{status.run.run_id} — started{' '} - {formatTime(status.run.start_time)} + Run #{status.run.runId} — started{' '} + {formatTime(status.run.startTime)}

)} @@ -232,27 +232,27 @@ export function JobsPage() { {runs.map((run) => ( - - {run.run_id} + + {run.runId} - {run.state?.life_cycle_state ?? '\u2014'} + {run.state?.lifeCycleState ?? '\u2014'} - {run.state?.result_state ?? '\u2014'} + {run.state?.resultState ?? '\u2014'} - {formatTime(run.start_time)} + {formatTime(run.startTime)} - {run.state?.life_cycle_state === 'RUNNING' && ( + {run.state?.lifeCycleState === 'RUNNING' && ( From 9ba517acb913388ff9a28f0c396024ebc6e56a59 Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Wed, 14 Oct 2026 13:00:11 +0200 Subject: [PATCH 3/3] refactor(appkit): type jobs runNow/submitRun results from SDK models Alias JobRunNowResponse/JobSubmitRunResponse as Numeric<> over the SDK's RunNowResponse/SubmitRunResponse instead of hand-written { runId: number }. Co-authored-by: Isaac Signed-off-by: MarioCadenas --- packages/appkit/src/connectors/jobs/client.ts | 6 ++++-- packages/appkit/src/connectors/jobs/index.ts | 1 + packages/appkit/src/connectors/jobs/types.ts | 4 ++++ packages/appkit/src/plugins/jobs/plugin.ts | 3 ++- packages/appkit/src/plugins/jobs/types.ts | 3 ++- packages/shared/src/workspace-client/modular.ts | 2 ++ 6 files changed, 15 insertions(+), 4 deletions(-) diff --git a/packages/appkit/src/connectors/jobs/client.ts b/packages/appkit/src/connectors/jobs/client.ts index 3b3aea1ac..dbf94ca71 100644 --- a/packages/appkit/src/connectors/jobs/client.ts +++ b/packages/appkit/src/connectors/jobs/client.ts @@ -24,8 +24,10 @@ import type { JobBaseRun, JobDetails, JobRun, + JobRunNowResponse, JobRunOutput, JobsConnectorConfig, + JobSubmitRunResponse, Numeric, } from "./types"; @@ -80,7 +82,7 @@ export class JobsConnector { workspaceClient: WorkspaceClient, request: SubmitRunRequest, signal?: AbortSignal, - ): Promise<{ runId: number }> { + ): Promise { return this._callApi("submit", async () => { const waiter = await workspaceClient.jobs.submitRun(request, { signal }); return { runId: Number(waiter.runId) }; @@ -91,7 +93,7 @@ export class JobsConnector { workspaceClient: WorkspaceClient, request: RunNowRequest, signal?: AbortSignal, - ): Promise<{ runId: number }> { + ): Promise { return this._callApi("runNow", async () => { const waiter = await workspaceClient.jobs.runNow(request, { signal }); return { runId: Number(waiter.runId) }; diff --git a/packages/appkit/src/connectors/jobs/index.ts b/packages/appkit/src/connectors/jobs/index.ts index 49359ab2c..4a31519fa 100644 --- a/packages/appkit/src/connectors/jobs/index.ts +++ b/packages/appkit/src/connectors/jobs/index.ts @@ -3,6 +3,7 @@ export type { JobBaseRun, JobDetails, JobRun, + JobRunNowResponse, JobRunOutput, JobsConnectorConfig, } from "./types"; diff --git a/packages/appkit/src/connectors/jobs/types.ts b/packages/appkit/src/connectors/jobs/types.ts index 01d255a92..e5b71d3d7 100644 --- a/packages/appkit/src/connectors/jobs/types.ts +++ b/packages/appkit/src/connectors/jobs/types.ts @@ -5,6 +5,8 @@ import type { GetJobResponse, GetRunOutputResponse, GetRunResponse, + RunNowResponse, + SubmitRunResponse, } from "../../workspace-client"; export interface JobsConnectorConfig { @@ -27,3 +29,5 @@ export type JobRun = Numeric; export type JobBaseRun = Numeric; export type JobRunOutput = Numeric; export type JobDetails = Numeric; +export type JobRunNowResponse = Numeric; +export type JobSubmitRunResponse = Numeric; diff --git a/packages/appkit/src/plugins/jobs/plugin.ts b/packages/appkit/src/plugins/jobs/plugin.ts index 9bc905929..084df58d2 100644 --- a/packages/appkit/src/plugins/jobs/plugin.ts +++ b/packages/appkit/src/plugins/jobs/plugin.ts @@ -12,6 +12,7 @@ import { type JobBaseRun, type JobDetails, type JobRun, + type JobRunNowResponse, type JobRunOutput, JobsConnector, } from "../../connectors/jobs"; @@ -287,7 +288,7 @@ class JobsPlugin extends Plugin { return { runNow: async ( params?: Record, - ): Promise> => { + ): Promise> => { const sdkFields = self._validateAndMap(jobKey, params); const result = await self.execute( diff --git a/packages/appkit/src/plugins/jobs/types.ts b/packages/appkit/src/plugins/jobs/types.ts index cfdae6b18..39989cb57 100644 --- a/packages/appkit/src/plugins/jobs/types.ts +++ b/packages/appkit/src/plugins/jobs/types.ts @@ -5,6 +5,7 @@ import type { JobBaseRun, JobDetails, JobRun, + JobRunNowResponse, JobRunOutput, } from "../../connectors/jobs"; import type { ExecutionResult } from "../../plugin"; @@ -40,7 +41,7 @@ export interface JobAPI { /** Trigger the configured job with validated params. Returns the run response. */ runNow( params?: Record, - ): Promise>; + ): Promise>; /** Trigger and poll until completion, yielding status updates. */ runAndWait( params?: Record, diff --git a/packages/shared/src/workspace-client/modular.ts b/packages/shared/src/workspace-client/modular.ts index 73553c24d..3449d4633 100644 --- a/packages/shared/src/workspace-client/modular.ts +++ b/packages/shared/src/workspace-client/modular.ts @@ -595,7 +595,9 @@ export type { GetRunResponse, ListRunsRequest, RunNowRequest, + RunNowResponse, SubmitRunRequest, + SubmitRunResponse, } from "@databricks/sdk-jobs/v2"; export type { EndpointHealth,