This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ecdfb67f2b3 TS SDK: Expose the Dag run conf on the task context
(#75563)
ecdfb67f2b3 is described below
commit ecdfb67f2b358130adf347533fd5ce2c6c88f354
Author: Guan-Ming Chiu <[email protected]>
AuthorDate: Sat Nov 7 14:22:15 2026 +0000
TS SDK: Expose the Dag run conf on the task context (#75563)
* TS SDK: Expose the Dag run conf on the task context
* Address review comments
Signed-off-by: Guan-Ming (Wesley) Chiu
<[email protected]>
---------
Signed-off-by: Guan-Ming (Wesley) Chiu
<[email protected]>
---
.../language-sdks/typescript.rst | 7 ++---
ts-sdk/src/coordinator/runtime.ts | 1 +
ts-sdk/src/sdk/task.ts | 16 +++++++++---
ts-sdk/tests/coordinator/client.test.ts | 1 +
ts-sdk/tests/coordinator/integration.test.ts | 30 ++++++++++++++++++++++
ts-sdk/tests/public-api.test.ts | 5 +++-
ts-sdk/tests/sdk/task-scope.test.ts | 1 +
7 files changed, 54 insertions(+), 7 deletions(-)
diff --git
a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
index d07fffcf9fc..3943fd6084b 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -552,9 +552,10 @@ Two functions give it what it needs while it runs:
- Returns
* - ``getContext()``
- The task's context (a ``TaskContext``): ``dagId``, ``taskId``
(including any task group prefix),
- ``runId``, ``tryNumber``, ``mapIndex`` (``-1`` for an unmapped task),
and ``signal``, an ``AbortSignal``
- that fires when Airflow stops the task. Pass ``signal`` to ``fetch()``,
timers, or any other API that
- accepts one, so the task can stop cleanly.
+ ``runId``, ``tryNumber``, ``mapIndex`` (``-1`` for an unmapped task),
``conf`` (the Dag run's
+ conf, ``{}`` when the run has none), and ``signal``, an ``AbortSignal``
that fires when Airflow stops the task.
+ Pass ``signal`` to ``fetch()``, timers, or any other API that accepts
one, so the task can stop cleanly.
+ ``getContext<T>()`` types ``conf`` as ``T``, as ``getXCom<T>`` does for
an XCom value.
* - ``getClient()``
- A ``TaskClient`` for Airflow Variables, Connections, XComs and the task
state store.
diff --git a/ts-sdk/src/coordinator/runtime.ts
b/ts-sdk/src/coordinator/runtime.ts
index b05a1b3c0ad..62ad38c7394 100644
--- a/ts-sdk/src/coordinator/runtime.ts
+++ b/ts-sdk/src/coordinator/runtime.ts
@@ -504,6 +504,7 @@ function buildContext(details: StartupDetails, signal:
AbortSignal): TaskContext
runId: details.ti.run_id,
tryNumber: details.ti.try_number,
mapIndex: details.ti.map_index ?? -1,
+ conf: (details.ti_context.dag_run?.conf ?? {}) as TaskContext["conf"],
signal,
};
}
diff --git a/ts-sdk/src/sdk/task.ts b/ts-sdk/src/sdk/task.ts
index 532afd9e04e..393e1e18d71 100644
--- a/ts-sdk/src/sdk/task.ts
+++ b/ts-sdk/src/sdk/task.ts
@@ -24,10 +24,11 @@
import { AsyncLocalStorage } from "node:async_hooks";
+import type { JsonValue } from "./client-types.js";
import type { TaskClient } from "./client.js";
/** Runtime metadata for the current task invocation. */
-export interface TaskContext {
+export interface TaskContext<TConf = Readonly<Record<string, JsonValue>>> {
/** Identifier of the Dag containing this task. */
readonly dagId: string;
/** Task ID for this handler invocation, including any TaskGroup prefix. */
@@ -38,6 +39,8 @@ export interface TaskContext {
readonly tryNumber: number;
/** -1 for non-mapped tasks, 0..N-1 for mapped instances. */
readonly mapIndex: number;
+ /** `conf` of the Dag run, or `{}` when the run has none. */
+ readonly conf: TConf;
/**
* AbortSignal that fires when Airflow terminates the task subprocess
* with SIGTERM or SIGINT.
@@ -100,10 +103,17 @@ function currentScope(accessor: ScopeAccessor): TaskScope
{
* }
* ```
*
+ * The generic `TConf` lets callers narrow `conf` when the shape is known:
+ *
+ * ```ts
+ * const { conf } = getContext<{ date: string }>();
+ * // conf is { date: string }
+ * ```
+ *
* @throws when called outside a task handler.
*/
-export function getContext(): TaskContext {
- return currentScope("getContext").ctx;
+export function getContext<TConf = TaskContext["conf"]>(): TaskContext<TConf> {
+ return currentScope("getContext").ctx as TaskContext<TConf>;
}
/**
diff --git a/ts-sdk/tests/coordinator/client.test.ts
b/ts-sdk/tests/coordinator/client.test.ts
index 01acff1eec9..a2133f785f4 100644
--- a/ts-sdk/tests/coordinator/client.test.ts
+++ b/ts-sdk/tests/coordinator/client.test.ts
@@ -40,6 +40,7 @@ const FAKE_CTX: TaskContext = {
runId: "r",
tryNumber: 1,
mapIndex: -1,
+ conf: {},
signal: new AbortController().signal,
};
diff --git a/ts-sdk/tests/coordinator/integration.test.ts
b/ts-sdk/tests/coordinator/integration.test.ts
index 4ba7b9bac53..e970138247d 100644
--- a/ts-sdk/tests/coordinator/integration.test.ts
+++ b/ts-sdk/tests/coordinator/integration.test.ts
@@ -286,6 +286,7 @@ describe("coordinator runtime integration", () => {
runId: "r1",
tryNumber: 1,
mapIndex: -1,
+ conf: {},
});
expect(result.logRecords.some((r) => r["event"] === "[ts-sdk.runtime] Task
succeeded")).toBe(
true,
@@ -302,6 +303,35 @@ describe("coordinator runtime integration", () => {
}
});
+ it("hands a handler the Dag run conf", async () => {
+ let observedConf: unknown = null;
+ testDag.task("read_conf", async () => {
+ observedConf = getContext<{ date: string; regions: string[] }>().conf;
+ })();
+
+ const conf = { date: "2026-01-01", regions: ["eu", "us"] };
+ const result = await driveSupervisor(
+ makeStartupDetails("read_conf", "test_dag", "r1", { dag_run: { conf } }),
+ );
+
+ expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" });
+ expect(observedConf).toEqual(conf);
+ });
+
+ it("hands a handler an empty conf when the Dag run has none", async () => {
+ let observedConf: unknown = undefined;
+ testDag.task("read_null_conf", async () => {
+ observedConf = getContext().conf;
+ })();
+
+ const result = await driveSupervisor(
+ makeStartupDetails("read_null_conf", "test_dag", "r1", { dag_run: {
conf: null } }),
+ );
+
+ expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" });
+ expect(observedConf).toEqual({});
+ });
+
it("times out when the terminal task response cannot be written", async ()
=> {
vi.useFakeTimers();
const originalWrite = NetSocket.prototype.write;
diff --git a/ts-sdk/tests/public-api.test.ts b/ts-sdk/tests/public-api.test.ts
index e5cf267dd77..36efb39d266 100644
--- a/ts-sdk/tests/public-api.test.ts
+++ b/ts-sdk/tests/public-api.test.ts
@@ -270,7 +270,9 @@ describe("public API", () => {
expectTypeOf<TaskFunction<{ regionCode: string },
number>>().toEqualTypeOf<
(args: { regionCode: string }) => number | Promise<number>
>();
- expectTypeOf<typeof getContext>().toEqualTypeOf<() => TaskContext>();
+ expectTypeOf<typeof getContext>().toEqualTypeOf<
+ <TConf = Readonly<Record<string, JsonValue>>>() => TaskContext<TConf>
+ >();
expectTypeOf<typeof getClient>().toEqualTypeOf<() => TaskClient>();
for (const name of ["TaskHandlerArgs", "runInTaskScope", "TaskScope"]) {
expect(name in sdk).toBe(false);
@@ -407,6 +409,7 @@ describe("public API", () => {
readonly runId: string;
readonly tryNumber: number;
readonly mapIndex: number;
+ readonly conf: Readonly<Record<string, JsonValue>>;
readonly signal: AbortSignal;
}>();
expectTypeOf<GetXComOpts>().toEqualTypeOf<{
diff --git a/ts-sdk/tests/sdk/task-scope.test.ts
b/ts-sdk/tests/sdk/task-scope.test.ts
index dc6299a4f6f..8526012564e 100644
--- a/ts-sdk/tests/sdk/task-scope.test.ts
+++ b/ts-sdk/tests/sdk/task-scope.test.ts
@@ -35,6 +35,7 @@ function makeScope(taskId: string): TaskScope {
runId: "r1",
tryNumber: 1,
mapIndex: -1,
+ conf: {},
signal: new AbortController().signal,
};
// Identity is all these tests read, so a cast beats a full fake client.
