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 9556dcf25b3 TS SDK: bind TaskFlow call arguments by folding names on
both sides (#73189)
9556dcf25b3 is described below
commit 9556dcf25b329a17a7a696648fabd3f716ef4501
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Fri Sep 18 21:41:16 2026 +0800
TS SDK: bind TaskFlow call arguments by folding names on both sides (#73189)
A Python Dag declares a task that runs in TypeScript with `@task.stub` and
calls it TaskFlow-style, but the handler could not see those arguments: it
was
called with nothing, so the call site was decoration and its values had to
be
hardcoded in the handler or re-fetched from XCom. Airflow already delivers
that
call site as an ordered, named spec in the task's run context.
The names are the problem worth solving. Python spells a parameter
`region_code` and TypeScript wants to read it as `regionCode`, and making an
author declare that for every ordinary snake_case parameter would be a tax
on
the common case. So names bind by folding on both sides, lowercased with
underscores removed, which is exactly the Go SDK's rule. One Python
signature
then binds identically in either SDK with nothing declared.
Folding on read rather than up front is why the bound object is a `Proxy`:
the
SDK sees Python's names and cannot know which spelling a handler will
destructure, so no guess at a TypeScript name is ever materialized. `in`
folds
like a read, and `Object.keys` and rest destructuring yield Python's names.
The
object has a null prototype, so a Python argument named `toString` binds
like
any other and one that was not passed misses rather than resolving to a
function.
An unmatched name logs rather than throws, since a destructuring default is
a
legitimate miss and nothing can tell one from a typo. The warning names both
the requested name and what the call delivered, and a failing task reports
the
same list, because a handler that destructured an argument under a name
nothing
folds to gets no error of its own.
Two Python names that fold to one token do fail the task, before the handler
runs and naming both: neither could be reached, and picking either silently
would hand the handler the wrong value. An XCom-backed binding fails the
same
way for now, with the getXCom call to write instead. Anything else the SDK
cannot honour fails rather than being dropped, because an unbound argument
reaches the handler as `undefined` and corrupts its output instead of
stopping
it.
---
.../language-sdks/typescript.rst | 43 +++-
.../ts_sdk_tests/test_ts_sdk_dag.py | 45 ++--
docs/spelling_wordlist.txt | 3 +
ts-sdk/README.md | 42 ++++
ts-sdk/docs/index.md | 15 ++
ts-sdk/example/dags/typescript_taskflow_example.py | 16 +-
ts-sdk/example/src/taskflow.ts | 57 ++++-
ts-sdk/src/coordinator/arg-binding.ts | 195 +++++++++++++++++
ts-sdk/src/coordinator/runtime.ts | 22 +-
ts-sdk/src/sdk/bundle.ts | 5 +-
ts-sdk/src/sdk/dag.ts | 4 +-
ts-sdk/src/sdk/task-handler.ts | 12 +-
ts-sdk/src/sdk/task.ts | 22 +-
ts-sdk/tests/coordinator/arg-binding.test.ts | 242 +++++++++++++++++++++
ts-sdk/tests/coordinator/integration.test.ts | 87 ++++++++
ts-sdk/tests/public-api.test.ts | 30 ++-
16 files changed, 784 insertions(+), 56 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 bf7c78bc6a7..6be73aa850c 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -131,11 +131,48 @@ through ``bundle.getTaskHandler(dagId, taskId)`` without
a coordinator runtime.
in progress. ``new Dag`` and ``dag.task`` take a trailing options object
(``spec`` on both, plus
``inputs`` on a task) that is not used yet; do not set them.
+TaskFlow arguments
+~~~~~~~~~~~~~~~~~~
+
+A Python Dag that calls a stub task TaskFlow-style passes those arguments
straight to the handler, which
+destructures them by name:
+
+.. code-block:: python
+
+ @task.stub(queue="typescript")
+ def transform(region_code: str, threshold: float, dry_run: bool = False):
...
+
+
+ transform("uk", 0.75)
+
+.. code-block:: typescript
+
+ interface TransformArgs {
+ regionCode: string;
+ threshold: number;
+ dryRun: boolean;
+ }
+
+ export async function transform({ regionCode, threshold, dryRun }:
TransformArgs) {
+ // ...
+ }
+
+Names bind by **folding on both sides**, lowercased with underscores removed,
so ``region_code`` reaches
+``regionCode`` and ``s3_uri`` reaches ``s3Uri`` with nothing declared on
either side.
+The Go SDK folds identically, so one Python signature binds the same way in
either SDK.
+An argument the call leaves at its default arrives carrying the default's
value.
+
+A name nothing folds to is **logged, not thrown**, naming what the handler
asked for and what the call
+delivered. Two Python names that fold to the same token fail the task.
+
+``Object.keys`` and rest destructuring (``{ ...rest }``) yield Python's names,
and ``in`` folds like a read.
+
.. note::
- As with the other language SDKs, XCom *dependencies* are declared in the
Python stub Dag (they define task
- order). The value must still be read explicitly in TypeScript via
``getClient().getXCom``, and produced
- either by the task's return value or by ``getClient().setXCom``.
+ An upstream's return value is not a bound argument unless the Python call
passes it. As with the other
+ language SDKs, XCom *dependencies* declared with ``>>`` in the Python stub
Dag define task order only.
+ Read a value the task was not passed explicitly via ``getClient().getXCom``,
and produce one either by
+ the task's return value or by ``getClient().setXCom``.
Coordinator configuration
~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git
a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py
index 5e25ef5801c..edd70b652d0 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py
@@ -21,17 +21,14 @@ Run with::
E2E_TEST_MODE=ts_sdk uv run --project airflow-e2e-tests pytest \\
tests/airflow_e2e_tests/ts_sdk_tests/ -xvs
-Two Dags mix Python tasks with ``@task.stub`` TypeScript tasks whose handlers
live in the single
-``airflow-ts-pack`` bundle built by ``conftest._setup_ts_sdk_integration``.
-Each is triggered once via a module-scoped fixture.
+Two Dags mix Python tasks with ``@task.stub`` TypeScript tasks, both served by
the single
+``airflow-ts-pack`` bundle, and each is triggered once via a module-scoped
fixture.
-``typescript_example`` confirms that ``NodeCoordinator`` launches the bundle
on the volume-provided
-Node runtime, that Variable/Connection reads and Python <-> TypeScript XCom
round-trips work through
-the Task Execution API, and that coordinator-channel logs reach the task-log
store.
+``typescript_example`` covers the runtime: Variable and Connection reads,
Python <-> TypeScript XCom
+round-trips, and task logs reaching the log store.
-``typescript_taskflow_example`` confirms that one bundle provides for two
``dag_id``s: its
-``build_message`` shares a ``task_id`` with a task in ``typescript_example``,
so a bundle that keyed
-dispatch on the task ID alone would run the wrong handler for one of them.
+``typescript_taskflow_example`` covers TaskFlow arguments, and shares a
``build_message`` task ID with
+``typescript_example`` so that dispatch keying on the task ID alone would run
the wrong handler.
"""
from __future__ import annotations
@@ -162,13 +159,31 @@ def
test_second_dag_from_the_same_bundle_succeeded(completed_taskflow_run: _Comp
)
-def test_summarize_xcom(completed_taskflow_run: _CompletedRun):
- """``summarize`` reads ``make_totals``'s output and averages it."""
+def test_summarize_binds_its_call_arguments(completed_taskflow_run:
_CompletedRun):
+ """Every argument the Dag's ``summarize("uk", "GBP", 280.0)`` call passes.
+
+ ``region_code`` and ``dry_run`` are snake_case in the ``@task.stub``
+ signature and camelCase in the handler, with nothing declared on either
+ side: folding is what carries them across. ``dry_run`` is left out of the
+ call, so it arrives from the stub's default.
+
+ A handler that received none of them would see ``undefined`` for each and
+ return nulls and ``NaN`` here rather than failing, which is why the whole
+ returned object is asserted.
+ """
assert completed_taskflow_run.xcom("make_totals") == {"orders": 12,
"revenue": 3402.0}
value = completed_taskflow_run.xcom("summarize")
- assert value == {"orders": 12, "averageOrder": 283.5, "currency": "GBP"}, (
- f"unexpected 'summarize' return_value: {value!r}"
- )
+ assert value == {
+ "regionCode": "uk",
+ "orders": 12,
+ "averageOrder": 283.5,
+ "currency": "GBP",
+ "passed": True,
+ "dryRun": False,
+ }, f"unexpected 'summarize' return_value: {value!r}"
+ # Written only when `dryRun` is false, so this also proves the defaulted
+ # boolean arrived as `false` rather than as `undefined`.
+ assert completed_taskflow_run.xcom("summarize", key="summary_line") ==
"uk: 12 orders"
def test_same_task_id_under_two_dags_runs_its_own_handler(
@@ -188,5 +203,5 @@ def test_same_task_id_under_two_dags_runs_its_own_handler(
)
assert taskflow_value == {
"dagId": _TASKFLOW_DAG_ID,
- "message": "12 orders averaging 283.5 GBP",
+ "message": "uk: 12 orders averaging 283.5 GBP",
}, f"unexpected 'typescript_taskflow_example.build_message' return_value:
{taskflow_value!r}"
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index ad18f656ec2..f35a04c3b16 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -490,6 +490,8 @@ deserializer
deserializes
deserializing
dest
+destructures
+destructuring
dev
devel
DevOps
@@ -1040,6 +1042,7 @@ logstash
longblob
lookups
loopback
+lowercased
lshift
lte
lxml
diff --git a/ts-sdk/README.md b/ts-sdk/README.md
index a4af960e622..013f31a3ed9 100644
--- a/ts-sdk/README.md
+++ b/ts-sdk/README.md
@@ -162,6 +162,48 @@ await new Bundle(
Register `TaskHandler` and `Dag` values with the `register` method, or pass
them to the `Bundle` constructor.
+## TaskFlow arguments
+
+A Python Dag that calls a stub task TaskFlow-style passes those arguments
straight to the handler,
+which destructures them by name:
+
+```python
+# the Python Dag
[email protected](queue="typescript")
+def transform(region_code: str, threshold: float, dry_run: bool = False): ...
+
+
+transform("uk", 0.75)
+```
+
+```ts
+interface TransformArgs {
+ regionCode: string;
+ threshold: number;
+ dryRun: boolean;
+}
+
+export async function transform({ regionCode, threshold, dryRun }:
TransformArgs) {
+ // ...
+}
+```
+
+Names bind by **folding on both sides**, lowercased with underscores removed,
so `region_code` reaches
+`regionCode` and `s3_uri` reaches `s3Uri` with nothing declared on either side.
+An argument the call leaves at its default arrives with the default's value.
+
+A name nothing folds to is **logged, not thrown**, naming what the handler
asked for and what the call
+delivered. Two Python names that fold to the same token fail the task.
+
+`Object.keys` and rest destructuring (`{ ...rest }`) yield Python's names, and
`in` folds like a read.
+
+An upstream's return value is not a bound argument unless the Python call
passes it.
+Read one the task was not passed explicitly:
+
+```ts
+const rows = await getClient().getXCom<number>({ key: "return_value", taskId:
"extract" });
+```
+
`Dag` is another interface, for a Dag declared natively in TypeScript, and is
still a work in progress.
Airflow launches the bundled entrypoint with `--comm=host:port` and
diff --git a/ts-sdk/docs/index.md b/ts-sdk/docs/index.md
index 689a8299e32..ab012a397d3 100644
--- a/ts-sdk/docs/index.md
+++ b/ts-sdk/docs/index.md
@@ -56,6 +56,21 @@ await bundle.serve();
`register` takes any number of items, so one bundle can provide for several
`TaskHandler`s.
+When the Python Dag calls a stub task TaskFlow-style, those arguments reach the
+handler by name. Names bind by folding on both sides, lowercased with
underscores removed,
+so a Python `region_code` reaches a handler's `regionCode` with nothing
declared:
+
+```ts
+interface TransformArgs {
+ regionCode: string;
+ threshold: number;
+}
+
+export async function transform({ regionCode, threshold }: TransformArgs) {
+ // ...
+}
+```
+
`Dag` is another interface, for a Dag declared in TypeScript rather than in
Python, and is still a work in progress.
## Coordinators
diff --git a/ts-sdk/example/dags/typescript_taskflow_example.py
b/ts-sdk/example/dags/typescript_taskflow_example.py
index 57b0d08ef9b..fbcc65fa4db 100644
--- a/ts-sdk/example/dags/typescript_taskflow_example.py
+++ b/ts-sdk/example/dags/typescript_taskflow_example.py
@@ -17,11 +17,11 @@
"""
-A second Python-owned Dag served by the same TypeScript bundle.
+TaskFlow argument binding across the language boundary.
-``typescript_example`` shows the basics; this Dag exists so one bundle
provides for two ``dag_id``s at once.
-Its ``build_message`` stub deliberately shares a ``task_id`` with a task in
``typescript_example``:
-a handler binds the ``(dag_id, task_id)`` pair, so the two are different tasks
with different bodies.
+``summarize`` is called TaskFlow-style, and every argument its call passes
reaches the TypeScript
+handler by name. Its ``build_message`` stub shares a ``task_id`` with a task
in ``typescript_example``
+on purpose: a handler binds the ``(dag_id, task_id)`` pair, so the two are
different tasks.
See ``src/taskflow.ts``.
"""
@@ -35,8 +35,12 @@ def make_totals():
return {"orders": 12, "revenue": 3402.0}
+# `region_code` and `dry_run` are snake_case on purpose: they reach the
handler's
+# `regionCode` and `dryRun` by folding, with nothing declared on either side.
+# The call below leaves `dry_run` at its default, and the handler receives that
+# value like any other.
@task.stub(queue="typescript")
-def summarize(): ...
+def summarize(region_code: str, currency: str, threshold: float, dry_run: bool
= False): ...
# Same task_id as `typescript_example.build_message`, on purpose.
@@ -51,7 +55,7 @@ def build_message(): ...
tags=["typescript", "example", "taskflow"],
)
def typescript_taskflow_example():
- make_totals() >> summarize() >> build_message()
+ make_totals() >> summarize("uk", "GBP", 280.0) >> build_message()
typescript_taskflow_example()
diff --git a/ts-sdk/example/src/taskflow.ts b/ts-sdk/example/src/taskflow.ts
index 54629cab87d..746665bf893 100644
--- a/ts-sdk/example/src/taskflow.ts
+++ b/ts-sdk/example/src/taskflow.ts
@@ -19,10 +19,13 @@
// Handlers for the `typescript_taskflow_example` Dag.
//
-// A second Python-owned Dag served by the same bundle, so the pair of ids a
handler binds is what
-// tells its tasks apart from `typescript_example`'s.
-// `build_summary_message` implements a task named `build_message`, exactly as
the other Dag has,
-// and the two share nothing else.
+// `summarize` shows argument binding. Its Python `@task.stub` signature is
snake_case and the
+// interface below is camelCase, and neither side declares anything, because
names bind by folding.
+//
+// The Dag is served by the same bundle as `typescript_example`, so the pair
of ids a handler binds
+// is what tells its tasks apart.
+// `buildSummaryMessage` implements a task named `build_message`, exactly as
the other Dag has, and
+// the two share nothing else.
import { getClient, getContext } from "apache-airflow-ts-sdk";
@@ -32,15 +35,40 @@ export interface Totals {
revenue: number;
}
+/**
+ * Every argument the Dag's `summarize(...)` call binds.
+ *
+ * Python spells these `region_code`, `currency`, `threshold` and `dry_run`.
+ * Folding absorbs the difference, so this interface says what the handler
+ * wants to read them as and nothing more.
+ */
+export interface SummarizeArgs {
+ regionCode: string;
+ currency: string;
+ threshold: number;
+ dryRun: boolean;
+}
+
/** What {@link summarize} returns, and what {@link buildSummaryMessage}
reads. */
export interface Summary {
+ regionCode: string;
orders: number;
averageOrder: number;
currency: string;
+ passed: boolean;
+ dryRun: boolean;
}
-export async function summarize(): Promise<Summary> {
- const totals = await getClient().getXCom<Totals>({
+export async function summarize({
+ regionCode,
+ currency,
+ threshold,
+ dryRun,
+}: SummarizeArgs): Promise<Summary> {
+ const client = getClient();
+ // Read explicitly: an upstream's return value is not a bound argument unless
+ // the Python call passes it, and this Dag's `summarize(...)` call does not.
+ const totals = await client.getXCom<Totals>({
key: "return_value",
taskId: "make_totals",
});
@@ -48,11 +76,22 @@ export async function summarize(): Promise<Summary> {
throw new Error(`task ${getContext().taskId} has no totals to summarize`);
}
const average = totals.orders === 0 ? 0 : totals.revenue / totals.orders;
+ const averageOrder = Number(average.toFixed(2));
+
+ if (!dryRun) {
+ await client.setXCom({
+ key: "summary_line",
+ value: `${regionCode}: ${totals.orders} orders`,
+ });
+ }
return {
+ regionCode,
orders: totals.orders,
- averageOrder: Number(average.toFixed(2)),
- currency: "GBP",
+ averageOrder,
+ currency,
+ passed: averageOrder >= threshold,
+ dryRun,
};
}
@@ -70,6 +109,6 @@ export async function buildSummaryMessage() {
// The dag_id is in the return value on purpose: it is how the end-to-end
test tells this task
// apart from `typescript_example`'s `build_message`.
dagId: ctx.dagId,
- message: `${summary.orders} orders averaging ${summary.averageOrder}
${summary.currency}`,
+ message: `${summary.regionCode}: ${summary.orders} orders averaging
${summary.averageOrder} ${summary.currency}`,
};
}
diff --git a/ts-sdk/src/coordinator/arg-binding.ts
b/ts-sdk/src/coordinator/arg-binding.ts
new file mode 100644
index 00000000000..49ff576e97b
--- /dev/null
+++ b/ts-sdk/src/coordinator/arg-binding.ts
@@ -0,0 +1,195 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// The TaskFlow call arguments Airflow captured for a task, delivered by name.
+//
+// A Python Dag declares a task that runs in another language with `@task.stub`
+// and calls it TaskFlow-style, as in `transform("uk", 0.75)`. Airflow
+// materializes that call site into an ordered, named spec and delivers it as
+// `StartupDetails.ti_context.arg_bindings`; this turns it back into the object
+// the handler destructures.
+//
+// Names bind by folding on both sides, so nothing has to be declared for a
+// Python `region_code` to reach a handler's `regionCode`.
+
+import type { LogChannel } from "./log-channel.js";
+import type { ArgBindings, TaskArgBinding } from "../generated/supervisor.js";
+import type { JsonValue } from "../sdk/client-types.js";
+
+/**
+ * Fold a name to the token both sides are matched on.
+ *
+ * Identical to the Go SDK's `strings.ToLower(strings.ReplaceAll(name, "_",
""))`,
+ * so one Python signature binds the same way in either SDK. Only `_` is
+ * removed, because it is the only separator a Python parameter name can
+ * contain.
+ */
+export function foldArgName(name: string): string {
+ return name.replaceAll("_", "").toLowerCase();
+}
+
+/** A task's TaskFlow call arguments, ready to hand to its handler. */
+export interface BoundArgs {
+ /**
+ * The object the handler is called with, a Proxy over the wire names:
reading
+ * a name folds it on demand, so `regionCode` and `region_code` both reach
+ * Python's `region_code`.
+ */
+ readonly args: object;
+ /**
+ * Python's names, in the calling signature's declaration order. What
+ * `Object.keys(args)` yields, and what a failing task reports.
+ */
+ readonly names: readonly string[];
+}
+
+const EMPTY_ARGS: BoundArgs = { args: Object.freeze({}), names:
Object.freeze([]) };
+
+/**
+ * Decode `ti_context.arg_bindings` into the object a task handler receives.
+ *
+ * A duplicate fold fails the task here, before the handler runs: two Python
+ * names that fold to one token cannot both be reached, and picking either
+ * silently would hand the handler the wrong value.
+ */
+export function bindArgs(bindings: ArgBindings | undefined, logs: LogChannel):
BoundArgs {
+ // Absent for an Airflow too old to send a spec, and for a task called with
no
+ // arguments. Both mean the handler's parameter has nothing in it.
+ if (bindings == null || bindings.length === 0) return EMPTY_ARGS;
+
+ const names: string[] = [];
+ const byFold = new Map<string, string>();
+ const values = new Map<string, JsonValue>();
+ for (const binding of bindings) {
+ const name = binding.name;
+ const fold = foldArgName(name);
+ const clash = byFold.get(fold);
+ if (clash !== undefined) {
+ throw new Error(
+ `Task arguments "${clash}" and "${name}" both fold to "${fold}", so a
handler cannot ` +
+ "name them apart; rename one in the @task.stub signature",
+ );
+ }
+ byFold.set(fold, name);
+ names.push(name);
+ values.set(name, resolveBindingValue(binding));
+ }
+
+ return { args: makeArgsProxy(names, byFold, values, logs), names };
+}
+
+/** The value a binding carries, or a throw for one this SDK cannot honour.
+ *
+ * A binding that cannot be honoured fails the task rather than being dropped:
+ * an unbound argument reaches the handler as `undefined`, which corrupts the
+ * task's output instead of stopping it. */
+function resolveBindingValue(binding: TaskArgBinding): JsonValue {
+ const { name } = binding;
+ if (binding.kind === "literal") {
+ // Airflow omits `value` for a literal whose value is null.
+ return (binding.value ?? null) as JsonValue;
+ }
+ if (binding.kind === "xcom") {
+ throw new Error(
+ `Task argument "${name}" takes the output of upstream task
"${binding.task_id}", but ` +
+ "XCom-backed arguments are not supported yet; read the value inside
the handler " +
+ `instead, with getClient().getXCom({ key: "return_value", taskId:
"${binding.task_id}" })`,
+ );
+ }
+ // Unreachable for the wire union as generated, so `binding` is `never` here.
+ // A newer Airflow can add a kind, and skipping it would silently leave the
+ // argument unbound, so the kind is read back off the value.
+ const kind: unknown = (binding as { kind?: unknown }).kind;
+ throw new Error(
+ `Task argument "${name}" has binding kind ${JSON.stringify(kind)}, which
this version ` +
+ "of apache-airflow-ts-sdk cannot bind; upgrade it to match this Airflow
release",
+ );
+}
+
+/**
+ * The object a handler destructures.
+ *
+ * A Proxy rather than a pre-built object with both spellings, because the SDK
+ * has no way to know which spelling a handler will use: it sees Python's names
+ * and nothing else. Folding on read means binding needs nothing declared on
+ * either side, and no guess about the TypeScript name is ever materialized.
+ */
+function makeArgsProxy(
+ names: readonly string[],
+ byFold: ReadonlyMap<string, string>,
+ values: ReadonlyMap<string, JsonValue>,
+ logs: LogChannel,
+): object {
+ const resolve = (property: string): string | undefined =>
+ values.has(property) ? property : byFold.get(foldArgName(property));
+
+ // A null prototype so a read never reaches Object.prototype: a Python
+ // argument named `constructor` or `toString` must bind like any other, and a
+ // handler destructuring one that was not passed must miss rather than get a
+ // function.
+ return new Proxy(Object.create(null) as Record<string, unknown>, {
+ get(_target, property) {
+ // Only string keys are arguments. A symbol read is something else, such
+ // as `Symbol.toPrimitive` during string coercion, and is not a miss.
+ if (typeof property !== "string") return undefined;
+ const name = resolve(property);
+ if (name !== undefined) return values.get(name);
+ // Logged, never thrown: a destructuring default such as
+ // `{ runId = "manual" }` is a legitimate miss, and nothing here can tell
+ // one from a typo.
+ logs.warning("Task argument not bound by this task's call", {
+ requested: property,
+ bound: [...names],
+ });
+ return undefined;
+ },
+ has(_target, property) {
+ // `in` folds like a read, so `"regionCode" in args` answers for the
+ // Python `region_code` the handler would actually receive.
+ return typeof property === "string" && resolve(property) !== undefined;
+ },
+ ownKeys() {
+ // Python's names: the SDK has no TypeScript-side names to enumerate, so
+ // `Object.keys` and rest destructuring (`{ ...rest }`) report what the
+ // wire actually delivered.
+ return [...names];
+ },
+ getOwnPropertyDescriptor(_target, property) {
+ if (typeof property !== "string" || !values.has(property)) return
undefined;
+ // Enumerable and configurable, or `ownKeys` would throw an invariant
+ // error for a key the target itself does not have.
+ return {
+ value: values.get(property),
+ writable: false,
+ enumerable: true,
+ configurable: true,
+ };
+ },
+ set(_target, property) {
+ throw new Error(
+ `Cannot assign to task argument ${String(property)}: bound arguments
are read-only`,
+ );
+ },
+ deleteProperty(_target, property) {
+ throw new Error(
+ `Cannot delete task argument ${String(property)}: bound arguments are
read-only`,
+ );
+ },
+ });
+}
diff --git a/ts-sdk/src/coordinator/runtime.ts
b/ts-sdk/src/coordinator/runtime.ts
index e34e37c1303..2087fc02e6d 100644
--- a/ts-sdk/src/coordinator/runtime.ts
+++ b/ts-sdk/src/coordinator/runtime.ts
@@ -36,6 +36,7 @@
// - DagFileParseRequest → respond with DagFileParsingResult, exit
// - StartupDetails → run task, respond Succeed or Fail, exit
//
+import { bindArgs, type BoundArgs } from "./arg-binding.js";
import { createCoordinatorClient } from "./client.js";
import { CommChannel } from "./comm-channel.js";
import { LogChannel } from "./log-channel.js";
@@ -307,15 +308,28 @@ async function handleTask(
const ctx = buildContext(details, signal);
const client = createCoordinatorClient(comm, ctx, clientLogs);
+
+ let bound: BoundArgs;
+ try {
+ bound = bindArgs(details.ti_context?.arg_bindings, logs);
+ } catch (err) {
+ // Before the handler ran, so nothing it might have written is at stake.
+ const message = (err as Error).message ?? String(err);
+ logs.error("Cannot bind this task's call arguments", {
+ task_id: ctx.taskId,
+ error: message,
+ });
+ return buildFailureResponse(details, message);
+ }
// Startup-details fields already logged above (`Received task
// startup details`); this line just marks the handler-call boundary.
- logs.debug("Dispatching to handler", { task_id: ctx.taskId });
+ logs.debug("Dispatching to handler", { task_id: ctx.taskId, bound_args:
bound.names });
try {
// The scope is installed around the call, not awaited inside it: the store
// follows the handler across every `await` it makes, so `getContext()` and
// `getClient()` work at any depth without the handler being handed either.
- const result = await runInTaskScope({ ctx, client }, () => handler());
+ const result = await runInTaskScope({ ctx, client }, () =>
handler(bound.args as never));
if (result !== undefined) {
await client.setXCom({ key: "return_value", value: result as JsonValue
});
}
@@ -335,6 +349,10 @@ async function handleTask(
task_id: ctx.taskId,
error: message,
stack: (err as Error).stack ?? null,
+ // A handler that destructured a bound argument under a name nothing
+ // folds to gets no error of its own, so a failing task says what its
+ // call actually delivered.
+ bound_args: bound.names,
});
return buildFailureResponse(details, message);
}
diff --git a/ts-sdk/src/sdk/bundle.ts b/ts-sdk/src/sdk/bundle.ts
index 612adbac8b7..ad99d0c8f34 100644
--- a/ts-sdk/src/sdk/bundle.ts
+++ b/ts-sdk/src/sdk/bundle.ts
@@ -32,7 +32,10 @@ let entriesOf: (bundle: Bundle) => ReadonlyMap<string,
BundleEntry>;
* {@link TaskHandler} for a task that a Python Dag declares, or a {@link Dag}
* declared in TypeScript.
*/
-export type Registerable = Dag | TaskHandler;
+// `never` for the handler's argument type: a TaskHandler is contravariant in
it
+// (the handler takes it), so this is the one instantiation every typed handler
+// is assignable to, `TaskHandler<TransformArgs>` included.
+export type Registerable = Dag | TaskHandler<never, unknown>;
// What a bundle holds per dag_id. The two arms are exclusive by construction:
// a Dag is the native case and owns its own tasks, while task handlers supply
diff --git a/ts-sdk/src/sdk/dag.ts b/ts-sdk/src/sdk/dag.ts
index 7374bf64883..1da149473bb 100644
--- a/ts-sdk/src/sdk/dag.ts
+++ b/ts-sdk/src/sdk/dag.ts
@@ -165,9 +165,9 @@ export class Dag {
* `taskId` must match the Dag-side operator's `task_id` exactly, including
* any TaskGroup prefix. Returns this task's handle.
*/
- task<TReturn = unknown>(
+ task<TArgs = void, TReturn = unknown>(
taskId: string,
- handler: TaskFunction<TReturn>,
+ handler: TaskFunction<TArgs, TReturn>,
options: TaskOptions = {},
): TaskRef {
if (typeof handler !== "function") {
diff --git a/ts-sdk/src/sdk/task-handler.ts b/ts-sdk/src/sdk/task-handler.ts
index a682b586d1b..0f29ea9f650 100644
--- a/ts-sdk/src/sdk/task-handler.ts
+++ b/ts-sdk/src/sdk/task-handler.ts
@@ -24,10 +24,10 @@ import { brand, hasBrand } from "./brand.js";
import type { TaskFunction } from "./task.js";
// Assigned inside TaskHandler's static block, as Dag does for its tasks.
-let functionOf: (handler: TaskHandler) => TaskFunction;
+let functionOf: (handler: TaskHandler<never, unknown>) => TaskFunction;
/** Internal: whether `value` is a TaskHandler built by any copy of this
package. */
-export function isTaskHandler(value: unknown): value is TaskHandler {
+export function isTaskHandler(value: unknown): value is TaskHandler<never,
unknown> {
return hasBrand(value, "TaskHandler");
}
@@ -62,18 +62,18 @@ function requireId(label: string, value: string): string {
* one the way a natively declared task is wired is a compile error rather than
* a runtime throw. For a native Dag, use {@link Dag} instead.
*/
-export class TaskHandler<TReturn = unknown> {
+export class TaskHandler<TArgs = void, TReturn = unknown> {
/** Identifier of the Python Dag this task belongs to. */
readonly dagId: string;
/** Airflow task ID this handler implements, including any TaskGroup prefix.
*/
readonly taskId: string;
- readonly #handler: TaskFunction<TReturn>;
+ readonly #handler: TaskFunction<TArgs, TReturn>;
static {
functionOf = (handler) => handler.#handler as TaskFunction;
}
- constructor(dagId: string, taskId: string, handler: TaskFunction<TReturn>) {
+ constructor(dagId: string, taskId: string, handler: TaskFunction<TArgs,
TReturn>) {
this.dagId = requireId("dagId", dagId);
this.taskId = requireId("taskId", taskId);
if (typeof handler !== "function") {
@@ -92,6 +92,6 @@ export class TaskHandler<TReturn = unknown> {
* private for the same reason `TaskRef` does not expose its handler: what a
* handler binds is identity, and reaching the body is the runtime's business.
*/
-export function getTaskHandlerFunction(handler: TaskHandler): TaskFunction {
+export function getTaskHandlerFunction(handler: TaskHandler<never, unknown>):
TaskFunction {
return functionOf(handler);
}
diff --git a/ts-sdk/src/sdk/task.ts b/ts-sdk/src/sdk/task.ts
index 3686691120b..532afd9e04e 100644
--- a/ts-sdk/src/sdk/task.ts
+++ b/ts-sdk/src/sdk/task.ts
@@ -131,10 +131,26 @@ export function getClient(): TaskClient {
/**
* Function signature for a TypeScript task handler.
*
- * A handler takes no SDK-supplied parameter: {@link getContext} and
- * {@link getClient} supply the runtime's side of the call.
+ * `TArgs` describes what the Dag's call site passes. {@link getContext} and
+ * {@link getClient} reach the runtime from inside the call, so neither is a
+ * parameter.
+ *
+ * ```ts
+ * interface TransformArgs {
+ * regionCode: string;
+ * threshold: number;
+ * }
+ *
+ * async function transform({ regionCode, threshold }: TransformArgs) {
+ * // ...
+ * }
+ * ```
+ *
+ * A handler that takes no arguments declares no parameter.
*
* Non-`undefined` return values are automatically pushed to XCom under the
* `"return_value"` key, matching Python `@task` behavior.
*/
-export type TaskFunction<TReturn = unknown> = () => TReturn | Promise<TReturn>;
+export type TaskFunction<TArgs = void, TReturn = unknown> = (
+ args: TArgs,
+) => TReturn | Promise<TReturn>;
diff --git a/ts-sdk/tests/coordinator/arg-binding.test.ts
b/ts-sdk/tests/coordinator/arg-binding.test.ts
new file mode 100644
index 00000000000..e08bf1e6401
--- /dev/null
+++ b/ts-sdk/tests/coordinator/arg-binding.test.ts
@@ -0,0 +1,242 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { describe, expect, it, vi } from "vitest";
+
+import { bindArgs, foldArgName, type BoundArgs } from
"../../src/coordinator/arg-binding.js";
+import type { LogChannel } from "../../src/coordinator/log-channel.js";
+import type { ArgBindings } from "../../src/generated/supervisor.js";
+
+function literal(name: string, value: unknown, extra: Record<string, unknown>
= {}) {
+ return { name, kind: "literal" as const, value, ...extra };
+}
+
+function makeLogs() {
+ const warning = vi.fn();
+ const logs = { warning } as unknown as LogChannel;
+ return { logs, warning };
+}
+
+function bind(bindings: ArgBindings): BoundArgs & { warning: ReturnType<typeof
vi.fn> } {
+ const { logs, warning } = makeLogs();
+ return { ...bindArgs(bindings, logs), warning };
+}
+
+describe("foldArgName", () => {
+ it.each([
+ ["region_code", "regioncode"],
+ ["regionCode", "regioncode"],
+ ["Name", "name"],
+ ["s3_uri", "s3uri"],
+ ["s3Uri", "s3uri"],
+ ["dry_run", "dryrun"],
+ ["already", "already"],
+ ["__dunder__", "dunder"],
+ ])("folds %s to %s", (name, folded) => {
+ expect(foldArgName(name)).toBe(folded);
+ });
+
+ it("matches the Go SDK's rule", () => {
+ // strings.ToLower(strings.ReplaceAll(name, "_", "")): only `_` is removed,
+ // because it is the only separator a Python parameter name can contain, so
+ // one Python signature binds identically in either SDK.
+ expect(foldArgName("a-b")).toBe("a-b");
+ });
+});
+
+describe("bindArgs", () => {
+ it("delivers nothing for a task called with no arguments", () => {
+ for (const bindings of [null, undefined, []] as (ArgBindings |
undefined)[]) {
+ const bound = bindArgs(bindings, makeLogs().logs);
+ expect(bound.names).toEqual([]);
+ expect(Object.keys(bound.args as object)).toEqual([]);
+ }
+ });
+
+ it("binds a camelCase name to Python's snake_case with nothing declared", ()
=> {
+ const { args } = bind([
+ literal("region_code", "uk"),
+ literal("threshold", 0.75),
+ literal("s3_uri", "s3://bucket/key"),
+ ]);
+ const { regionCode, threshold, s3Uri } = args as {
+ regionCode: string;
+ threshold: number;
+ s3Uri: string;
+ };
+
+ expect({ regionCode, threshold, s3Uri }).toEqual({
+ regionCode: "uk",
+ threshold: 0.75,
+ s3Uri: "s3://bucket/key",
+ });
+ });
+
+ it("binds in the other direction too, and for a capitalised name", () => {
+ // Folding is symmetric, so a Python side that already uses camelCase or a
+ // capitalised name needs nothing declared either.
+ const { args } = bind([literal("regionCode", "uk"), literal("Name",
"United Kingdom")]);
+ const { region_code: regionCode, name } = args as { region_code: string;
name: string };
+
+ expect({ regionCode, name }).toEqual({ regionCode: "uk", name: "United
Kingdom" });
+ });
+
+ it("binds an exact name without folding it", () => {
+ const { args } = bind([literal("threshold", 0.75)]);
+ expect((args as { threshold: number }).threshold).toBe(0.75);
+ });
+
+ it("binds every JSON value a literal can carry", () => {
+ const { args } = bind([
+ literal("totals", { orders: 12, revenue: 3402 }),
+ literal("regions", ["uk", "de"]),
+ literal("dry_run", false),
+ literal("retries_used", 3),
+ // Airflow omits `value` entirely for a literal whose value is null.
+ { name: "label", kind: "literal" as const },
+ ]);
+
+ expect({ ...(args as object) }).toEqual({
+ totals: { orders: 12, revenue: 3402 },
+ regions: ["uk", "de"],
+ dry_run: false,
+ retries_used: 3,
+ label: null,
+ });
+ });
+
+ it("binds an argument the call left at its default", () => {
+ // Arrives flagged `from_default`, which changes nothing about the value:
+ // the handler cannot tell, and should not need to.
+ const { args } = bind([literal("dry_run", true, { from_default: true })]);
+ expect((args as { dryRun: boolean }).dryRun).toBe(true);
+ });
+
+ it("logs an unmatched name rather than throwing", () => {
+ // A destructuring default such as `{ runId = "manual" }` is a legitimate
+ // miss, and nothing can tell one from a typo, so a miss cannot fail a
task.
+ const { args, warning } = bind([literal("region_code", "uk")]);
+ const { runId = "manual", reigonCode } = args as { runId?: string;
reigonCode?: string };
+
+ expect(runId).toBe("manual");
+ expect(reigonCode).toBeUndefined();
+ expect(warning).toHaveBeenCalledWith("Task argument not bound by this
task's call", {
+ requested: "runId",
+ bound: ["region_code"],
+ });
+ // Both the requested name and what the call actually delivered, so a typo
+ // is diagnosable from the task log alone.
+ expect(warning).toHaveBeenCalledWith("Task argument not bound by this
task's call", {
+ requested: "reigonCode",
+ bound: ["region_code"],
+ });
+ });
+
+ it("does not log a symbol read as an unbound argument", () => {
+ // Promise resolution, string coercion and test frameworks all probe an
+ // object with symbols; none of those is an argument that went missing.
+ const { args, warning } = bind([literal("region_code", "uk")]);
+ void (args as Record<symbol, unknown>)[Symbol.toPrimitive];
+ void (args as Record<symbol, unknown>)[Symbol.iterator];
+
+ expect(warning).not.toHaveBeenCalled();
+ });
+
+ it("folds `in` like a read", () => {
+ const { args } = bind([literal("region_code", "uk")]);
+
+ expect("regionCode" in (args as object)).toBe(true);
+ expect("region_code" in (args as object)).toBe(true);
+ expect("threshold" in (args as object)).toBe(false);
+ });
+
+ it("yields Python's names from Object.keys and rest destructuring", () => {
+ // The SDK has no TypeScript-side names to enumerate: it sees the wire's
+ // names and nothing else, so that is what enumeration reports.
+ const bindings: ArgBindings = [literal("region_code", "uk"),
literal("dry_run", false)];
+ const { args, names } = bind(bindings);
+ const { ...rest } = args as object;
+
+ expect(names).toEqual(["region_code", "dry_run"]);
+ expect(Object.keys(args as object)).toEqual(["region_code", "dry_run"]);
+ expect(rest).toEqual({ region_code: "uk", dry_run: false });
+ expect(Object.entries(args as object)).toEqual([
+ ["region_code", "uk"],
+ ["dry_run", false],
+ ]);
+ });
+
+ it("keeps the declaration order of the calling signature", () => {
+ const { names } = bind([literal("c", 1), literal("a", 2), literal("b",
3)]);
+ expect(names).toEqual(["c", "a", "b"]);
+ });
+
+ it("fails at dispatch when two Python names fold to the same token", () => {
+ // Neither could be reached by name, and picking either silently would hand
+ // the handler the wrong value.
+ expect(() => bind([literal("region_code", "uk"), literal("regionCode",
"de")])).toThrowError(
+ /Task arguments "region_code" and "regionCode" both fold to
"regioncode"/,
+ );
+ });
+
+ it("does not reach Object.prototype for an argument that was not passed", ()
=> {
+ // A Python argument named `constructor` or `toString` must bind like any
+ // other, and a handler destructuring one that was not passed must miss.
+ const { args } = bind([literal("toString", "not a function")]);
+
+ expect((args as { toString: unknown }).toString).toBe("not a function");
+ expect((args as { constructor?: unknown }).constructor).toBeUndefined();
+ expect("valueOf" in (args as object)).toBe(false);
+ });
+
+ it("binds an argument named __proto__ as a key", () => {
+ const { args } = bind([literal("__proto__", { polluted: true })]);
+
+ expect(Object.keys(args as object)).toEqual(["__proto__"]);
+ expect(({} as { polluted?: boolean }).polluted).toBeUndefined();
+ });
+
+ it("refuses an XCom-backed argument, naming the upstream task", () => {
+ expect(() =>
+ bind([{ name: "totals", kind: "xcom" as const, task_id: "make_totals"
}]),
+ ).toThrowError(/takes the output of upstream task "make_totals"/);
+ });
+
+ it("refuses a binding kind from a newer Airflow", () => {
+ // Skipping it would leave the argument unbound, and an unbound argument
+ // destructures to `undefined` and corrupts the task's output.
+ const unknownKind = [{ name: "totals", kind: "dataset" }] as unknown as
ArgBindings;
+ expect(() => bind(unknownKind)).toThrowError(
+ /has binding kind "dataset", which this version of apache-airflow-ts-sdk
cannot bind/,
+ );
+ });
+
+ it("refuses assignment and deletion", () => {
+ // The bound object mirrors a call site that already happened, so writing
+ // to it would change nothing an author could observe downstream.
+ const { args } = bind([literal("region_code", "uk")]);
+
+ expect(() => {
+ (args as { regionCode: string }).regionCode = "de";
+ }).toThrowError(/bound arguments are read-only/);
+ expect(() => {
+ delete (args as { region_code?: string }).region_code;
+ }).toThrowError(/bound arguments are read-only/);
+ });
+});
diff --git a/ts-sdk/tests/coordinator/integration.test.ts
b/ts-sdk/tests/coordinator/integration.test.ts
index fabd39a089c..51f87c4c780 100644
--- a/ts-sdk/tests/coordinator/integration.test.ts
+++ b/ts-sdk/tests/coordinator/integration.test.ts
@@ -40,6 +40,14 @@ import { Bundle } from "../../src/sdk/bundle.js";
import { TaskHandler } from "../../src/sdk/task-handler.js";
import { getClient, getContext } from "../../src/sdk/task.js";
+/** The arguments `py_dag.bound`'s Python call site passes, as its handler
+ * spells them. Folding absorbs the snake_case on the wire. */
+interface BoundTransformArgs {
+ regionCode: string;
+ threshold: number;
+ dryRun: boolean;
+}
+
const testDag = new Dag("test_dag");
const otherDag = new Dag("other_dag");
// The bundle the runtime dispatches through. startCoordinator() is driven
@@ -403,6 +411,85 @@ describe("coordinator runtime integration", () => {
}
});
+ it("hands a handler the arguments its Dag's call bound", async () => {
+ // End of the folding path through the real wire format: the Python call
+ // site's names arrive snake_case and the handler destructures camelCase.
+ let observed: unknown = null;
+ bundle.register(
+ new TaskHandler(
+ "py_dag",
+ "bound",
+ async ({ regionCode, threshold, dryRun }: BoundTransformArgs) => {
+ observed = { regionCode, threshold, dryRun };
+ return observed;
+ },
+ ),
+ );
+
+ const result = await driveSupervisor(
+ makeStartupDetails("bound", "py_dag", "r1", {
+ arg_bindings: [
+ { name: "region_code", kind: "literal", value: "uk" },
+ { name: "threshold", kind: "literal", value: 0.75 },
+ { name: "dry_run", kind: "literal", value: false, from_default: true
},
+ ],
+ }),
+ );
+
+ expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" });
+ expect(observed).toEqual({ regionCode: "uk", threshold: 0.75, dryRun:
false });
+ });
+
+ it("fails the task when two of its bound names fold alike", async () => {
+ // Reported before the handler runs, so nothing it might have written to
+ // XCom is at stake.
+ bundle.register(new TaskHandler("py_dag", "ambiguous", async () => "never
runs"));
+
+ const result = await driveSupervisor(
+ makeStartupDetails("ambiguous", "py_dag", "r1", {
+ arg_bindings: [
+ { name: "region_code", kind: "literal", value: "uk" },
+ { name: "regionCode", kind: "literal", value: "de" },
+ ],
+ }),
+ );
+
+ expect(result.firstResponse!.body).toMatchObject({ type: "TaskState",
state: "failed" });
+ expect(result.runtimeRequests.filter((r) => r.type ===
"SetXCom")).toHaveLength(0);
+ expect(
+ result.logRecords.some(
+ (r) =>
+ r["event"] === "[ts-sdk.runtime] Cannot bind this task's call
arguments" &&
+ String(r["error"]).includes('both fold to "regioncode"'),
+ ),
+ ).toBe(true);
+ });
+
+ it("names what a failing task's call bound", async () => {
+ // A handler that destructured an argument under a name nothing folds to
+ // gets no error of its own, so the failure report carries the names.
+ bundle.register(
+ new TaskHandler("py_dag", "misnamed", async () => {
+ throw new Error("cannot proceed");
+ }),
+ );
+
+ const result = await driveSupervisor(
+ makeStartupDetails("misnamed", "py_dag", "r1", {
+ arg_bindings: [{ name: "region_code", kind: "literal", value: "uk" }],
+ }),
+ );
+
+ expect(result.firstResponse!.body).toMatchObject({ type: "TaskState",
state: "failed" });
+ expect(
+ result.logRecords.some(
+ (r) =>
+ r["event"] === "[ts-sdk.runtime] Task failed" &&
+ JSON.stringify(r["bound_args"]) === JSON.stringify(["region_code"]),
+ ),
+ ).toBe(true);
+ });
+
it("aborts the context signal on SIGTERM and reports a thrown task error",
async () => {
let sawAbort = false;
testDag.task("aborted_then_failed", async () => {
diff --git a/ts-sdk/tests/public-api.test.ts b/ts-sdk/tests/public-api.test.ts
index dae26146e1a..ccb37542700 100644
--- a/ts-sdk/tests/public-api.test.ts
+++ b/ts-sdk/tests/public-api.test.ts
@@ -164,9 +164,17 @@ describe("public API", () => {
// Identity and a body, and no more: no schedule, no task order, no dag_id
// of its own to declare.
expectTypeOf<keyof TaskHandler>().toEqualTypeOf<"dagId" | "taskId">();
- expectTypeOf<ConstructorParameters<typeof TaskHandler>>().toEqualTypeOf<
- [string, string, TaskFunction<unknown>]
- >();
+ expectTypeOf<TaskHandler["dagId"]>().toEqualTypeOf<string>();
+ expectTypeOf<TaskHandler["taskId"]>().toEqualTypeOf<string>();
+
+ // The handler's own parameter type is inferred, so a typed handler needs
+ // no type argument written out at the registration site.
+ const typed = new TaskHandler(
+ "py_etl",
+ "report",
+ async ({ regionCode }: { regionCode: string }) =>
regionCode.toUpperCase(),
+ );
+ expectTypeOf(typed).toEqualTypeOf<TaskHandler<{ regionCode: string },
string>>();
});
it("does not let a task handler be wired the way a native task is", () => {
@@ -196,9 +204,13 @@ describe("public API", () => {
});
it("are the only way a handler reaches the runtime", () => {
- // A handler is a plain function of its own data, so the SDK hands it no
- // parameter at all and the scope is not something an author installs.
- expectTypeOf<TaskFunction>().toEqualTypeOf<() => unknown |
Promise<unknown>>();
+ // A handler is a plain function of its own data: the parameter carries
+ // the Dag's arguments and nothing else, and the scope is not something
+ // an author installs.
+ expectTypeOf<TaskFunction>().toEqualTypeOf<(args: void) => unknown |
Promise<unknown>>();
+ expectTypeOf<TaskFunction<{ regionCode: string },
number>>().toEqualTypeOf<
+ (args: { regionCode: string }) => number | Promise<number>
+ >();
expectTypeOf<typeof getContext>().toEqualTypeOf<() => TaskContext>();
expectTypeOf<typeof getClient>().toEqualTypeOf<() => TaskClient>();
for (const name of ["TaskHandlerArgs", "runInTaskScope", "TaskScope"]) {
@@ -226,7 +238,7 @@ describe("public API", () => {
expectTypeOf<Bundle["serve"]>().toEqualTypeOf<() => Promise<void>>();
expectTypeOf<Bundle["register"]>().toEqualTypeOf<(...items:
Registerable[]) => void>();
expectTypeOf<ConstructorParameters<typeof
Bundle>>().toEqualTypeOf<Registerable[]>();
- expectTypeOf<Registerable>().toEqualTypeOf<Dag | TaskHandler>();
+ expectTypeOf<Registerable>().toEqualTypeOf<Dag | TaskHandler<never,
unknown>>();
for (const name of ["serveDags", "DagRegistry"]) {
expect(name in sdk).toBe(false);
}
@@ -247,9 +259,9 @@ describe("public API", () => {
}>();
expectTypeOf<ConstructorParameters<typeof Dag>>().toEqualTypeOf<[string,
DagSpec?]>();
expectTypeOf<Dag["task"]>().toEqualTypeOf<
- <TReturn = unknown>(
+ <TArgs = void, TReturn = unknown>(
taskId: string,
- handler: TaskFunction<TReturn>,
+ handler: TaskFunction<TArgs, TReturn>,
options?: TaskOptions,
) => TaskRef
>();
