TS SDK: bind TaskFlow call arguments by folding names on both sides

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
 ~~~~~~~~~~~~~~~~~~~~~~~~~

airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py

@@ -21,17 +21,14 @@
     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}"

docs/spelling_wordlist.txt

@@ -489,6 +489,8 @@ deserializer
 deserializes
 deserializing
 dest
+destructures
+destructuring
 dev
 devel
 DevOps
@@ -1038,6 +1040,7 @@ logstash
 longblob
 lookups
 loopback
+lowercased
 lshift
 lte
 lxml

ts-sdk/README.md

@@ -163,6 +163,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
+@task.stub(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

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

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()

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,27 +35,63 @@ 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",
   });
   if (totals === null) {
     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}`,
   };
 }