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 6e5d9ddc185 Update the TypeScript SDK docs for Dags declared in
TypeScript (#73875)
6e5d9ddc185 is described below
commit 6e5d9ddc18526b4b465272341fbd8810d0faf94e
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Fri Oct 9 14:23:13 2026 +0800
Update the TypeScript SDK docs for Dags declared in TypeScript (#73875)
* Update the TypeScript SDK docs for Dags declared in TypeScript
Describe the TypeScript SDK as it will be once the native Dag authoring
surface and native Dag parsing land: declaring a whole Dag in TypeScript,
and implementing mixed-language tasks, from one guide.
* Match the TypeScript SDK docs to the APIs on main
dag.if and dag.switch take the deciding handler and its inputs, and
triggerDagRun goes through dag.task and runs on Node.js, so Jinja in conf
is not rendered. The guide also picks up dag_bundle_to_coordinator, UTC
cron schedules, .airflowignore for handler-only bundles and the CLI
commands that refuse a TypeScript Dag. The overview marks Go as able to
declare Dags.
* Call a Dag declared in a language SDK a Dag definition
"Native Dag" reads as jargon. Dag definition names the SDK feature, as a
reviewer suggested.
* Call mixed-language tasks a Python Dag with stub TaskHandler
A task is only ever written in one language: a Python Dag declares it as
a stub, and a TaskHandler in the SDK's language implements it.
---------
---
.../language-sdks/index.rst | 81 +-
.../language-sdks/typescript.rst | 882 ++++++++++-----------
ts-sdk/README.md | 283 ++-----
ts-sdk/docs/index.md | 68 +-
ts-sdk/example/README.md | 23 +-
5 files changed, 574 insertions(+), 763 deletions(-)
diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst
index 159911d84c9..8e6e725b999 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst
@@ -22,31 +22,40 @@ Non-Python Task SDKs
|experimental|
-Airflow Dags are always defined in Python, but individual task
*implementations* can be written in other
-languages. When a task runs, Airflow's worker calls out the target language to
execute the task logic, which
-communicates the result back. The Dag author uses a lightweight Python *stub*
to declare where the task lives;
-everything else, including the actual business logic, Airflow API calls, and
any library dependencies, lives
-in a non-Python implementation.
+Airflow Dags and tasks can be written in languages other than Python. A
language SDK is used in one of two
+ways:
+
+* **Dag definition.** The whole Dag, with its schedule, tasks and
dependencies, is defined in the SDK's language.
+ See :ref:`language-sdks/dag-definition`.
+* **Python Dag with stub TaskHandler.** A Python Dag declares the tasks and
their dependencies, and a
+ TaskHandler in the SDK's language implements each stub task. See
:ref:`language-sdks/stub-taskhandler`.
+
+Either way, a task runs on a *coordinator*, which starts the language's
runtime for the task and relays
+messages between it and Airflow. See :ref:`language-sdks/coordinator-config`.
.. list-table:: Available language SDKs
:header-rows: 1
- :widths: 15 40 15 30
+ :widths: 15 35 15 15 20
* - Language
- Coordinator class
- Min. runtime
+ - Dag definition
- Guide
* - JVM languages (e.g. Java)
- :class:`task-sdk:airflow.sdk.coordinators.java.JavaCoordinator`
- JRE 17
+ - Yes
- :doc:`java`
* - Go
-
:class:`task-sdk:airflow.sdk.coordinators.executable.ExecutableCoordinator`
- None (native binary)
+ - Yes
- :doc:`go`
* - TypeScript
- :class:`task-sdk:airflow.sdk.coordinators.node.NodeCoordinator`
- Node.js 22
+ - Yes
- :doc:`typescript`
.. toctree::
@@ -56,36 +65,38 @@ in a non-Python implementation.
go
typescript
-How it works
-------------
+.. _language-sdks/dag-definition:
+
+Dag definition
+--------------
+
+The whole Dag is declared in the SDK's language: its schedule, its tasks,
their options and the dependencies
+between them. No Python file is involved.
-The execution model has three moving parts.
+The SDK builds the Dag into an artifact, and the artifact goes into a Dag
bundle like any other Dag file. The
+Dag processor runs the artifact with the language's runtime to read its Dags,
so the Dag processor needs that
+runtime and the ``[sdk]`` coordinator configuration. From there, Airflow
treats the Dag like any Python Dag:
+the scheduler plans its runs, it shows up in the UI, and each task runs on the
coordinator its queue maps to,
+from the same Dag bundle.
-**Stub tasks in the Dag**
- The Dag file declares tasks using :func:`@task.stub
<airflow.sdk.task.stub>`. A stub is a normal Airflow
- task from the scheduler's perspective. It participates in dependencies,
retries, pools, and all other
- task-level features exactly like other ``@task``-decorated Python
functions. The only difference is that
- the worker does not execute the Python code inside the function definition;
instead it delegates execution
- to a *coordinator*.
+The **Code** view in the Airflow UI shows the source the SDK embeds in the
artifact.
-**Coordinators**
- A coordinator is a Python object registered in the ``[sdk] coordinators``
configuration. This is considered
- a part of an Airflow worker. When the worker picks up a stub task, it looks
up the coordinator mapped to
- that task's specified ``queue``, and uses the coordinator to execute the
task. The coordinator is
- responsible for managing the target language's runtime, forwarding messages
from, and relaying results back
- to Airflow. All coordinators extend
- :class:`task-sdk:airflow.sdk.execution_time.coordinator.BaseCoordinator`.
+For TypeScript, see :ref:`typescript-sdk/dag-definition`.
-**Language runtime**
- The coordinator calls out one short-lived runtime per task instance. In
most cases, this would be a
- subprocess of an executable implemented in a non-Python language. The
runtime receives messages from the
- worker to identify the workload, executes the task, and communicates
through the coordinator as a proxy
- back to the worker process.
+.. _language-sdks/stub-taskhandler:
+
+Python Dag with stub TaskHandler
+--------------------------------
+
+A Python Dag declares a task that another language implements as a *stub
task*. The scheduler sees a normal
+task: it takes part in dependencies, retries, pools and every other task-level
feature like any other
+``@task`` function. When a worker picks it up, the worker does not run the
Python function body; it hands the
+task to the coordinator its queue maps to, which runs the implementation in
the target language.
.. _language-sdks/stub-tasks:
Stub tasks
-----------
+~~~~~~~~~~
A stub task is declared with the :func:`@task.stub <airflow.sdk.task.stub>`
decorator. Since it is still a
Python task declaration, every parameter available on a normal Dag or task
applies. Task dependencies are also
@@ -129,16 +140,22 @@ documentation on how to do this correctly.
.. note::
- For a Dag containing stub tasks, the **Code** view in the Airflow UI shows
only the Python Dag
- file — including the stub declarations — as the Dag's source. The
non-Python implementation source
- is not displayed anywhere in the UI; consult your project repository or
the build artifact shipped
- in the bundle to inspect it. This is an intentional architecture decision,
not a bug.
+ For a Dag containing stub tasks, the **Code** view in the Airflow UI shows
only the Python Dag file,
+ including the stub declarations, as the Dag's source. The non-Python
implementation source is not
+ displayed anywhere in the UI; consult your project repository or the build
artifact shipped in the bundle
+ to inspect it. This is an intentional architecture decision, not a bug.
.. _language-sdks/coordinator-config:
Coordinator configuration
-------------------------
+A coordinator is a Python object registered in the ``[sdk] coordinators``
configuration, and runs as part of an
+Airflow worker. When the worker picks up a task, it looks up the coordinator
mapped to the task's ``queue``,
+and uses it to run the task. The coordinator starts one short-lived runtime
per task instance, usually a
+subprocess of an executable written in the target language, and relays
messages between that runtime and the
+worker. All coordinators extend
:class:`task-sdk:airflow.sdk.execution_time.coordinator.BaseCoordinator`.
+
Coordinators are registered in ``airflow.cfg`` (or via environment variables)
under ``[sdk]``.
``coordinators``
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 a3589d6019a..d2c6094a105 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -22,12 +22,17 @@ TypeScript SDK
|experimental|
-The TypeScript SDK lets you register task handlers on a ``Bundle`` and
implement their logic in TypeScript (or
-plain JavaScript), running on Node.js. A matching Python stub Dag still
declares the scheduling shape and
-dependencies; individual tasks delegate to a Node.js subprocess that is
spawned by
-:class:`~airflow.sdk.coordinators.node.NodeCoordinator` for each task instance.
+The TypeScript SDK lets you write Airflow Dags and tasks in TypeScript, or
plain JavaScript, and run them on
+Node.js. There are two ways to use it:
-The SDK is the ``apache-airflow-ts-sdk`` package (ESM-only). It is currently
in **beta** and its API may change.
+* **Dag definition.** The schedule, the tasks, their options and the
dependencies between them are all written
+ in TypeScript, with no Python file involved. See
:ref:`typescript-sdk/dag-definition`.
+* **Python Dag with stub TaskHandler.** A Python Dag declares the tasks with
``@task.stub`` and wires them
+ together, and a TypeScript ``TaskHandler`` implements each one. See
:ref:`typescript-sdk/stub-tasks`.
+
+Both use the same task API and the same build tool, and one bundle can serve
both.
+
+The SDK is the ``apache-airflow-ts-sdk`` npm package (ESM-only). It is in
**beta**, and its API may change.
.. warning::
@@ -37,9 +42,9 @@ The SDK is the ``apache-airflow-ts-sdk`` package (ESM-only).
It is currently in
.. seealso::
- For the full TypeScript API reference (``Bundle``, ``TaskHandler``, ``Dag``,
the task handler getters,
- ``TaskClient``, supporting types, and exceptions),
- see the `TypeScript SDK API reference
<https://airflow.apache.org/docs/ts-sdk/stable/>`__.
+ For the full API reference (``Dag``, ``Bundle``, ``TaskHandler``,
``withArgNames``, the task getters,
+ ``TaskClient``, supporting types, and exceptions), see the
+ `TypeScript SDK API reference
<https://airflow.apache.org/docs/ts-sdk/stable/>`__.
.. contents:: Contents
:local:
@@ -48,519 +53,496 @@ The SDK is the ``apache-airflow-ts-sdk`` package
(ESM-only). It is currently in
Prerequisites
-------------
-* Node.js 22 or later must be available on the Airflow worker nodes and the
Dag processor.
- Once a ``NodeCoordinator`` is configured, the Dag processor runs ``node`` on
every packed ``*.min.mjs`` bundle
- in every Dag bundle, including ones that only register ``TaskHandler``
objects,
- and reports an import error for each when ``node`` is missing.
-* The packed bundle (a single ``bundle.min.mjs`` file, see
:ref:`typescript-sdk/build`) must be accessible
- from the worker and the Dag processor, in the Dag bundle the coordinator
scans.
-* The ``apache-airflow-task-sdk`` package (installed with Airflow) provides
the coordinator; no additional
- Python packages are needed.
-* In the TypeScript project, install the ``apache-airflow-ts-sdk`` npm package
to author task handlers:
+* Node.js 22 or later on the Airflow workers. A Dag declared in TypeScript
also needs Node.js on the Dag
+ processor, which runs the bundle to read the Dag. To add Node.js to the
Airflow image, extend it as
+ described in :doc:`docker-stack:build`. If ``node`` is not on the ``PATH``,
set the coordinator's
+ ``node_executable`` (see :ref:`typescript-sdk/coordinator-config`).
+* The ``apache-airflow-task-sdk`` package, installed with Airflow, provides
+ :class:`~airflow.sdk.coordinators.node.NodeCoordinator`, which runs the
TypeScript code. No other Python
+ package is needed.
+* In your TypeScript project, install the SDK, and ``esbuild`` to build the
bundle:
.. code-block:: bash
npm install apache-airflow-ts-sdk
+ npm install --save-dev esbuild
+
+.. _typescript-sdk/quick-start:
Quick start
-----------
-The following example shows the minimal moving parts: a Python Dag with a stub
task, and a TypeScript
-implementation of that task.
+This example declares a whole Dag in TypeScript, builds it into a bundle, and
deploys it, with no Python
+file involved.
-Python Dag (the scheduling side)
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Declare a Dag in ``src/main.ts``:
-.. code-block:: python
+.. code-block:: typescript
- from airflow.sdk import dag, task
+ import { Bundle, Dag } from "apache-airflow-ts-sdk";
+ const dag = new Dag("ts_hello", { schedule: "@daily", queue: "typescript"
});
- @dag
- def typescript_example():
- @task
- def python_start():
- return "hello from Python"
+ const extract = dag.task("extract", async (): Promise<number> => 42);
+ const report = dag.task("report", async ({ rows }: { rows: number }) => {
+ console.log(`extracted ${rows} rows`);
+ });
- @task.stub(queue="typescript")
- def build_message(): ...
+ report({ rows: extract() });
- python_start() >> build_message()
+ await new Bundle(dag).serve();
+Build it into a single file:
- typescript_example()
+.. code-block:: bash
-``@task.stub`` declares the *shape* of the TypeScript task without any Python
implementation. The ``queue``
-value routes the task to the Node.js coordinator.
+ npx airflow-ts-pack src/main.ts --outdir dist
-TypeScript implementation
-~~~~~~~~~~~~~~~~~~~~~~~~~
+Make sure ``dist/bundle.min.mjs`` is in a Dag bundle, for example the default
``dags-folder`` bundle, which
+reads the ``[core] dags_folder`` directory (see :doc:`Dag bundles
</administration-and-deployment/dag-bundles>`).
+Then send the ``typescript`` queue to the Node.js coordinator in
``airflow.cfg``:
-A task is an ordinary (usually ``async``) function taking no arguments:
-``getContext()`` and ``getClient()`` reach the runtime from inside the call,
so nothing the SDK supplies is a parameter.
+.. code-block:: ini
-Create a ``TaskHandler`` per task, binding the function to the ``dag_id`` and
``task_id`` it implements,
-register them on a ``Bundle``, then serve it to Airflow with
``bundle.serve()``.
-That top-level ``await`` makes the module a runnable bundle entry point.
+ [sdk]
+ coordinators = {"ts": {"classpath":
"airflow.sdk.coordinators.node.NodeCoordinator"}}
+ queue_to_coordinator = {"typescript": "ts"}
-.. code-block:: typescript
+Airflow reads the bundle like any other Dag file: ``ts_hello`` shows up in the
UI, runs on its schedule,
+and ``report`` receives the value ``extract`` returned.
- import { Bundle, getClient, TaskHandler } from "apache-airflow-ts-sdk";
+.. _typescript-sdk/dag-definition:
- export async function buildMessage() {
- const client = getClient();
- const upstream = await client.getXCom<string>({
- key: "return_value",
- taskId: "python_start",
- });
- const greeting = await client.getVariable("typescript_example_greeting");
- return `${greeting ?? "hello from TypeScript"}; upstream=${upstream ??
"missing"}`;
- }
+Declaring a Dag in TypeScript
+-----------------------------
- const bundle = new Bundle();
- bundle.register(new TaskHandler("typescript_example", "build_message",
buildMessage));
- await bundle.serve();
+Tasks and their inputs
+~~~~~~~~~~~~~~~~~~~~~~
-The ``dagId`` a handler binds must match the ``dag_id`` of the Python Dag, and
the ``taskId`` a
-``@task.stub`` function in that Dag, including any TaskGroup prefix.
+``dag.task(taskId, handler)`` declares a task and returns a *factory*. Calling
the factory places the task in
+the Dag and gives it its inputs, so the calls you write are the dependencies:
-``register`` takes any number of task handlers and Dags, and
``bundle.serve()`` serves exactly what is
-registered, so a task left out is not part of the packed bundle and is marked
removed at runtime.
-A second ``bundle.serve()`` call is rejected.
-Registering holds no sockets and starts nothing, so a unit test can build a
bundle and dispatch a handler
-through ``bundle.getTaskHandler(dagId, taskId)`` without a coordinator runtime.
+.. code-block:: typescript
-TaskFlow arguments
-~~~~~~~~~~~~~~~~~~
+ import { Dag } from "apache-airflow-ts-sdk";
-A Python Dag that calls a stub task TaskFlow-style passes those arguments
straight to the handler, which
-destructures them by name:
+ const dag = new Dag("ts_etl");
-.. code-block:: python
+ const extract = dag.task("extract", async (): Promise<number> => 42);
+ const transform = dag.task(
+ "transform",
+ async ({ rows, region }: { rows: number; region: string }) => rows * 2,
+ );
+ const load = dag.task("load", async ({ total }: { total: number }) => {});
- @task.stub(queue="typescript")
- def transform(region_code: str, threshold: float, dry_run: bool = False):
...
+ const extracted = extract();
+ const total = transform({ rows: extracted, region: "us" });
+ load({ total });
+A handler takes one object of named arguments, and a call names each input. An
input is either the
+reference another task's call returned, which makes this task wait for that
task and receive its value, or a
+literal JSON value such as ``"us"``. A task with no arguments is called with
none, and a single argument is
+named like any other: ``load({ total })``.
- transform("uk", 0.75)
+The compiler checks every call: an argument left out, a misspelled one, and a
literal of the wrong type are
+all compile errors.
+
+The argument type can be a named interface, and the handler a function
declared anywhere, including in another
+module. This is also how a task passes data to the next one:
.. code-block:: typescript
- interface TransformArgs {
- regionCode: string;
- threshold: number;
- dryRun: boolean;
+ interface Summary {
+ total: number;
+ regions: number;
}
- export async function transform({ regionCode, threshold, dryRun }:
TransformArgs) {
- // ...
+ interface ReportArgs {
+ summary: Summary;
+ label: string;
+ }
+
+ export async function report({ summary, label }: ReportArgs) {
+ console.log(`${label}: ${summary.total} rows in ${summary.regions}
regions`);
}
-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.
+ const summarize = dag.task("summarize", async (): Promise<Summary> => ({
total: 42, regions: 2 }));
+ const reportTask = dag.task("report", report);
-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.
+ reportTask({ summary: summarize(), label: "nightly" });
-``Object.keys`` and rest destructuring (``{ ...rest }``) yield Python's names,
and ``in`` folds like a read.
+``report`` receives the object ``summarize`` returned as ``summary``, and the
literal ``"nightly"`` as
+``label``. Values move between tasks as XComs, so a task returns data JSON can
hold (see
+:ref:`typescript-sdk/types`).
-An argument the call fills from another task, as in ``transform(extract(),
"uk")``,
-arrives as that task's value rather than a reference to it.
-An upstream that pushed no output fails the task, naming both the argument and
the task it came from;
-one that pushed ``null`` binds ``null``.
+A few rules keep the graph honest:
-A Python ``int`` beyond the ±9007199254740991 a JavaScript number holds
exactly is refused rather than
-bound, so carry such a value across the language boundary as a string.
+* Every task is called exactly once. A task that is never called fails when
the Dag is read, so none is left
+ out of the graph by accident.
+* A reference has to be the input itself. One placed inside an array or an
object is treated as a literal and
+ draws no dependency.
-Explicit renames
-~~~~~~~~~~~~~~~~
+Task ids
+~~~~~~~~
-``withArgNames`` states a binding when folding cannot reach it, for a name the
Python side never used:
-a clearer word than the Dag chose, or a TypeScript reserved word like ``enum``.
-The mapping comes first, the handler second:
+The task id may be left out, in which case it is the handler's function name:
.. code-block:: typescript
- interface ReportArgs {
- summary: Summary;
- label: string; // Python calls this `run_label`
- }
-
- const report = withArgNames({ label: "run_label" }, async ({ summary,
label }: ReportArgs) => {
- // `label` is the call's `run_label`; `summary` folded as usual.
+ const extract = dag.task(async function extract(): Promise<number> {
+ return 42;
});
- bundle.register(new TaskHandler("etl", "report", report));
-
-An entry beats folding, and everything the map does not mention still folds,
-so ``withArgNames`` should be rare in a real Dag.
-A mapped name the call did not pass misses rather than falling back to folding.
+``airflow-ts-pack`` keeps function names, so building the bundle never renames
a task. A handler with no name
+of its own, such as an arrow function written inline, needs an id: pass it
first, or set ``taskId`` in the
+task's options. Give it in one place only.
-The map's keys are checked against the handler's own parameter type, so ``{
labl: "run_label" }`` is a
-compile error naming the right key. Its values are Python names, which ``tsc``
cannot see and does not
-check.
+A task id you write is made of letters, digits, dashes and underscores.
-.. note::
+Dag and task options
+~~~~~~~~~~~~~~~~~~~~
- Being upstream is not the same as being passed. As with the other language
SDKs, an XCom *dependency*
- declared with ``>>`` in the Python stub Dag defines task order only. Read a
value the call did not pass
- explicitly via ``getClient().getXCom``, and produce one either by the task's
return value or by
- ``getClient().setXCom``.
+``new Dag`` and ``dag.task`` both take a trailing object of Airflow options.
The fields are Airflow's own,
+spelled in camelCase, and a misspelled one is a compile error:
-Coordinator configuration
-~~~~~~~~~~~~~~~~~~~~~~~~~
+.. code-block:: typescript
-Register a Dag bundle for the packed bundles, register the coordinator, and
route the queue to it in
-``airflow.cfg`` (or the equivalent ``AIRFLOW__*`` environment variables):
+ const dag = new Dag("ts_etl", {
+ schedule: "@daily",
+ catchup: false,
+ tags: ["etl"],
+ queue: "typescript",
+ });
-.. code-block:: ini
+ const load = dag.task("load", loadRows, { retries: 2, retryDelay: 30 });
- [dag_processor]
- dag_bundle_config_list = [
- {"name": "dags-folder", "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
- {
- "name": "ts-task-handlers",
- "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
- "kwargs": {"path": "/opt/airflow/ts-bundles"}
- }
- ]
+``schedule`` accepts ``@once``, ``@continuous``, a cron expression, or a cron
preset such as ``@daily``. Leave
+it out for a Dag that only runs when triggered. A cron schedule runs in UTC.
- [sdk]
- coordinators = {
- "ts": {
- "classpath": "airflow.sdk.coordinators.node.NodeCoordinator",
- "kwargs": {"task_handler_bundle_name": "ts-task-handlers"}
- }
- }
- queue_to_coordinator = {"typescript": "ts"}
+Every task of the Dag runs on the Node.js coordinator, so it needs a queue that
+:ref:`queue_to_coordinator <typescript-sdk/coordinator-config>` sends there.
Set ``queue`` once on the Dag and
+each task inherits it. A task's own ``queue`` wins over the Dag's.
-``task_handler_bundle_name`` names the Dag bundle the coordinator scans for
packed bundles;
-``queue_to_coordinator`` routes stub tasks with ``queue="typescript"`` to this
coordinator. See
-:ref:`typescript-sdk/coordinator-config` for the full list of accepted
``kwargs`` and how the bundle is
-located.
+Order-only dependencies
+~~~~~~~~~~~~~~~~~~~~~~~
-There is no separate Node.js worker to run: the Airflow worker launches the
bundle with ``node`` once per
-task instance.
+When a task only has to run after another, without receiving its value, draw
the dependency between the
+references with ``before`` and ``after``, the TypeScript spelling of Python's
``>>`` and ``<<``:
-A Dag processor with this ``[sdk]`` configuration also parses the
``*.min.mjs`` bundles of every Dag bundle,
-and needs Node.js to do so (see :ref:`typescript-sdk/native-parsing`).
-The ``ts-task-handlers`` Dag bundle only holds the bundles that the Python
stub Dag's tasks run,
-so keep the Dag processor from parsing it by listing ``*`` in its
``.airflowignore``:
+.. code-block:: typescript
-.. code-block:: bash
+ const loaded = load({ total });
+ const cleaned = cleanup();
- echo '*' > /opt/airflow/ts-bundles/.airflowignore
+ loaded.before(cleaned); // loaded >> cleaned
+ cleaned.after(loaded, extracted); // [loaded, extracted] >> cleaned
-.. note::
+Both take any number of references, and drawing a dependency that already
exists changes nothing. Each returns
+the reference it was called on, so ``loaded.before(cleaned).before(notified)``
draws both from ``loaded``.
- The ``[sdk]`` config, the packed ``*.min.mjs`` bundles and Node.js must be
present wherever tasks execute
- and on the Dag processor. With ``CeleryExecutor``, tasks execute on the
Celery workers; with
- ``LocalExecutor``, they run inside the scheduler process. The Dag processor
checks the stub tasks of each
- Python Dag against the task handlers the packed bundles register, so it runs
them too.
- It also runs them to parse the bundles of every Dag bundle, see
:ref:`typescript-sdk/native-parsing`.
- The API server does not need any of it.
- Register the Dag bundle in ``[dag_processor] dag_bundle_config_list`` on
every
- component, like your other Dag bundles: the worker and the Dag processor
resolve
- ``task_handler_bundle_name`` through it, and wherever the ``[sdk]`` config
is read it is rejected if the
- name is missing there.
+Pass a value as an input when the downstream task needs it, and use ``before``
or ``after`` when it only needs
+to run in order.
-.. _typescript-sdk/native-dag:
+Task groups
+~~~~~~~~~~~
-Declaring a Dag in TypeScript
------------------------------
+``dag.taskGroup(groupId)`` opens a group with the same ``task`` and
``taskGroup`` methods as the Dag. Every id
+declared in it is prefixed with the group's, as in Python:
-A ``Dag`` is declared on this side rather than in Python: its schedule, its
tasks, their options and
-the edges between them are all written in TypeScript. Airflow parses such a
Dag from its packed bundle,
-see :ref:`typescript-sdk/native-parsing`.
+.. code-block:: typescript
-``dag.task(taskId, handler)`` returns a *factory*. A handler takes one object
of named arguments, and
-calling the factory names each input, so the call graph is the task graph:
+ const staging = dag.taskGroup("staging");
+ const staged = staging.task("stage_rows", stageRows)(); // task
id "staging.stage_rows"
+ staging.taskGroup("checks").task("nulls", checkNulls)(); // task
id "staging.checks.nulls"
-.. code-block:: typescript
+ staging.before(loaded); //
staging >> loaded
- import { Dag } from "apache-airflow-ts-sdk";
+A group takes ``before`` and ``after`` too, which orders the whole group
against a task or another group.
- const dag = new Dag("ts_etl");
+Tasks and groups share one id namespace, so a Dag cannot hold both a task and
a group called ``staging``. The
+``.`` is added by the group, so an id you write cannot contain one.
- const extract = dag.task("extract", async (): Promise<number> => 42);
- const transform = dag.task(
- "transform",
- async ({ rows, region }: { rows: number; region: string }) => rows * 2,
- );
- const load = dag.task("load", async ({ total }: { total: number }) => {});
+Pass ``{ prefixGroupId: false }`` to keep the ids declared in a group as
written, as ``prefix_group_id=False``
+does in Python; they then have to be unique across the Dag. A group id is made
of letters, digits, dashes and
+underscores, and is at most 200 characters.
- const extracted = extract();
- const total = transform({ rows: extracted, region: "us" });
- load({ total });
+Conditional branching
+~~~~~~~~~~~~~~~~~~~~~
-Naming the inputs is how a task is called. A handler that takes no arguments
is called with none, and
-a single argument is named like any other, ``load({ total })``. The compiler
checks the call: it reports
-an argument left out, a misspelled one, and a literal of the wrong type.
+``dag.if`` declares a task from a handler that returns a boolean, and names
the task each outcome runs:
-Each argument takes either an upstream reference or a literal JSON value. A
reference has to be the
-argument itself: one buried inside an array or an object is a literal, and
draws no edge.
+.. code-block:: typescript
-Every task has to be called exactly once. An uncalled task fails when the Dag
is read, so none can be
-left out of the graph by accident.
+ async function hasRows({ total }: { total: number }): Promise<boolean> {
+ return total > 0;
+ }
-The task id may be omitted, in which case it is the handler's function name:
+ dag.if(hasRows, { total }).then(loaded).else(reportedEmpty);
-.. code-block:: typescript
+Its inputs come second, and an optional third argument takes task options,
such as ``taskId``; the id is
+otherwise the handler's name. The compiler checks that the handler returns a
boolean. ``else`` is optional:
+without it, the ``then`` task is skipped when the condition is false.
- const extract = dag.task(async function extract(): Promise<number> {
- return 42;
- });
+The tasks a condition guards take no input from it. To use a value the
condition computed, read it with
+``getClient().getXCom``.
-``airflow-ts-pack`` keeps function names intact, so bundling cannot rename a
task. A handler with no
-name of its own, such as an arrow function passed inline, has nothing to take
an id from and needs
-one: either positionally or as ``taskId`` in its spec. Give it in one place
only, not both.
+The side not taken is skipped, and stays skipped if it is cleared later. Only
the tasks named here are
+skipped, so a task that both sides lead to still runs. This differs from
Python's ``@task.branch``, which
+skips every direct downstream task it did not choose.
-Order-only edges
-~~~~~~~~~~~~~~~~
+Multi-way branching
+~~~~~~~~~~~~~~~~~~~
-An edge that carries no value has no argument name to travel under, so it is
drawn between the
-references themselves with ``before`` and ``after``, the TypeScript pair for
Python's ``>>`` and
-``<<``:
+``dag.switch`` declares a task from a handler that returns the reference of
the task to run, and lists the
+candidates:
.. code-block:: typescript
- const loaded = load({ transformed });
- const cleaned = cleanup();
+ const daily = publishDaily();
+ const weekly = publishWeekly();
- loaded.before(cleaned); // loaded >> cleaned
- cleaned.after(loaded, transformed); // [loaded, transformed] >>
cleaned
+ async function pickCadence() {
+ const cadence = await getClient().getVariable("cadence");
+ return cadence === "weekly" ? weekly : daily;
+ }
-Both take any number of references, so one call draws several edges, and
drawing an edge that
-already exists changes nothing. Each returns the reference it was called on, so
-``loaded.before(cleaned).before(notified)`` draws both edges from ``loaded``.
+ dag.switch(pickCadence).case(daily).case(weekly);
-Pass a value as an argument when the downstream task needs it, and use
``before`` or ``after`` when
-it only needs to run in order.
+It takes inputs and options as ``dag.if`` does. A case is a task reference, so
the compiler checks it exists,
+and renaming a handler cannot silently change which task runs. The deciding
task's value is the id of the
+task it chose.
-Task groups
-~~~~~~~~~~~
-
-``dag.taskGroup(groupId)`` opens a scope with the same ``task`` and
``taskGroup`` methods as the Dag,
-prefixing the id of everything declared in it, as Python's ``prefix_group_id``
does:
+Exactly one case runs, and there is no default: a decider that returns
anything else fails, naming what it
+returned and what it could have returned. To run several tasks together on one
outcome, put them behind a
+single task, or give each its own ``dag.if``.
-.. code-block:: typescript
+Triggering another Dag
+~~~~~~~~~~~~~~~~~~~~~~
- const staging = dag.taskGroup("staging");
- staging.task("stage_rows", stageRows)(); // task id
"staging.stage_rows"
- staging.taskGroup("checks").task("nulls", checkNulls)(); //
"staging.checks.nulls"
+``triggerDagRun`` makes a task that starts a run of another Dag. Pass it to
``dag.task`` in place of a handler:
- staging.before(loaded); // staging >> loaded
+.. code-block:: typescript
-A group is an edge endpoint in its own right, so ``before`` and ``after``
order a whole group against
-a task or against another group.
+ import { triggerDagRun } from "apache-airflow-ts-sdk";
-Tasks and groups share one id namespace, as they do in Python, so a Dag cannot
hold both a task and a
-group called ``staging``. A ``.`` is what separates a group from what it
holds, so it cannot appear in
-an id of either.
+ const triggered = dag.task(
+ "trigger_downstream",
+ triggerDagRun({ dagId: "downstream_etl", conf: { source: "ts_etl" },
waitForCompletion: true }),
+ )();
-Pass ``{ prefixGroupId: false }`` to keep the ids declared in a group as
written, as ``prefix_group_id=False``
-does in Python; they then have to be unique across the Dag. A group id is made
of letters, digits, dashes and
-underscores, and is at most 200 characters.
+ triggered.after(loaded);
-Serialization
-~~~~~~~~~~~~~
+The task has no handler to take an id from, so it needs one, and it takes no
inputs. ``triggerDagRun`` takes
+the options of Python's ``TriggerDagRunOperator``, such as ``conf``,
``waitForCompletion``, ``deferrable``,
+``pokeInterval``, ``allowedStates`` and ``failedStates``. Values are sent as
written, so Jinja in ``conf`` is
+not rendered.
-A native Dag serializes into the same Dag JSON a Python Dag produces, so the
scheduler reads it
-without knowing which language declared it.
+The task runs on Node.js like the Dag's other tasks, and inherits the Dag's
``queue``. With ``deferrable``, it
+waits in the triggerer, which needs the standard provider installed.
-``schedule`` accepts what maps to a stock timetable: unset, ``@once``,
``@continuous``, or a cron
-expression. A cron preset such as ``@daily`` is recorded as the expression it
stands for. Anything
-else names a Python object a TypeScript bundle cannot point at, and is
rejected.
+A complete example
+~~~~~~~~~~~~~~~~~~
-A cron schedule is read in UTC. Python takes the Dag's timezone from
``start_date``, but
-``startDate`` here is a ``Date``, which is an instant and carries no timezone,
so there is none to
-take.
+``ts-sdk/example/src/native.ts`` uses all of the above in one Dag: a task
group, a fan-in, a condition, a
+multi-way branch, order-only dependencies and a triggered Dag run.
-Every task of a native Dag runs on the Node coordinator, so it needs the queue
the deployment routes
-there. Set it once on the Dag and each task inherits it:
+.. _typescript-sdk/stub-tasks:
-.. code-block:: typescript
+Implementing stub tasks of a Python Dag
+---------------------------------------
- const dag = new Dag("ts_etl", { schedule: "@daily", queue: "typescript" });
+Here a Python Dag declares the tasks and their dependencies, and TypeScript
implements some of them. Use it
+when the Dag needs something a TypeScript Dag cannot express yet (see
:ref:`typescript-sdk/limitations`), or to
+mix Python and TypeScript tasks in one Dag.
- // ...and one task that needs its own.
- dag.task("heavy", heavyHandler, { queue: "typescript_large" })();
+The Python Dag
+~~~~~~~~~~~~~~
-``queue`` on a task wins over the Dag's. See
:ref:`typescript-sdk/coordinator-config` for the
-``queue_to_coordinator`` entry that sends that queue to the coordinator.
+.. code-block:: python
-Conditional branching
-~~~~~~~~~~~~~~~~~~~~~
+ from airflow.sdk import dag, task
-``dag.if`` takes a handler that returns a boolean, and names the task each
outcome runs:
-.. code-block:: typescript
+ @dag
+ def typescript_example():
+ @task
+ def python_start():
+ return "hello from Python"
- async function hasRows({ rows }: { rows: number }): Promise<boolean> {
- return rows > 0;
- }
+ @task.stub(queue="typescript")
+ def build_message(): ...
- const gate = dag.if(hasRows, { rows: extracted });
- gate.then(loaded).else(reportedEmpty);
+ python_start() >> build_message()
-``dag.if`` declares the condition as a task: its id is the function's name,
and a trailing spec sets
-its id or other options, as in ``dag.if(hasRows, { rows }, { taskId:
"has_rows" })``. The compiler
-checks that the handler returns a boolean. ``else`` is optional: a one-sided
condition skips its own
-branch when the condition fails and follows nothing. The condition is also a
node, so
-``notified.after(gate)`` orders a task after it.
-A guarded task takes no argument for the control edge, because a condition's
boolean decides whether
-the task runs rather than what it runs on. Read a value from the condition with
-``getClient().getXCom``.
+ typescript_example()
-The side not taken is skipped when the run reaches it, and stays skipped if
you clear it later. Only
-the branches named here are skipped, so a task that several branches converge
on still runs — unlike
-Python's ``@task.branch``, which skips every immediate downstream it did not
follow.
+``@task.stub`` declares a task that is implemented elsewhere, and ``queue``
sends it to the Node.js
+coordinator.
-Multi-way branching
+The TypeScript side
~~~~~~~~~~~~~~~~~~~
-``dag.switch`` is the multi-way form: a handler that returns one of the cases
it is given.
+Create a ``TaskHandler`` per stub task, naming the ``dag_id`` and ``task_id``
it implements, and register them
+on a ``Bundle``:
.. code-block:: typescript
- async function pickPath({ rows }: { rows: number }): Promise<TaskRef> {
- return rows > 1000 ? handleLong : handleShort;
+ import { Bundle, getClient, TaskHandler } from "apache-airflow-ts-sdk";
+
+ export async function buildMessage() {
+ const client = getClient();
+ const upstream = await client.getXCom<string>({ key: "return_value",
taskId: "python_start" });
+ const greeting = await client.getVariable("typescript_example_greeting");
+ return `${greeting ?? "hello from TypeScript"}; upstream=${upstream ??
"missing"}`;
}
- dag.switch(pickPath, { rows: extracted
}).case(handleLong).case(handleShort);
+ const bundle = new Bundle();
+ bundle.register(new TaskHandler("typescript_example", "build_message",
buildMessage));
+ await bundle.serve();
+
+The ``taskId`` must match the stub's task id, including any task group prefix.
-``dag.switch`` declares the decider the way ``dag.if`` does. A case is the
task reference itself, so
-the compiler checks the candidate exists and renaming a handler cannot
silently rewire a Dag. The
-task's own value is the chosen task's id, which a downstream task can read
from its XCom.
+``register`` takes any number of task handlers and Dags, so one bundle can
implement the tasks of several
+Python Dags and declare Dags of its own. A task left out of the bundle is
marked removed when it runs.
-There is no default case. A decider that returns anything outside its cases
fails the task, naming
-what it chose and what it could have chosen.
+Nothing connects to Airflow until ``bundle.serve()``, so a unit test can build
a bundle and call a handler
+through ``bundle.getTaskHandler(dagId, taskId)`` without running Airflow.
-Exactly one case is selected. Python's branch callable may return a list of
task ids, and no language
-SDK offers that yet: put the paths that run together behind one task, or gate
each with its own
-condition.
+TaskFlow arguments
+~~~~~~~~~~~~~~~~~~
-Triggering another Dag
-~~~~~~~~~~~~~~~~~~~~~~
+A Python Dag that calls a stub task TaskFlow-style passes those arguments 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):
...
-``triggerDagRun`` starts another Dag's run. Pass it to ``dag.task`` in place
of a handler:
+
+ transform("uk", 0.75)
.. code-block:: typescript
- import { triggerDagRun } from "apache-airflow-ts-sdk";
+ interface TransformArgs {
+ regionCode: string;
+ threshold: number;
+ dryRun: boolean;
+ }
+
+ export async function transform({ regionCode, threshold, dryRun }:
TransformArgs) {
+ // ...
+ }
+
+Names match ignoring case and underscores, so ``region_code`` reaches
``regionCode`` and ``s3_uri`` reaches
+``s3Uri`` with nothing declared on either side. The Go SDK matches names the
same way. An argument the call
+leaves at its default arrives with the default's value.
+
+A name the handler reads that the call did not pass is logged rather than
raised, naming what the handler
+asked for and what the call delivered. Two Python names that match the same
TypeScript name fail the task.
- const trigger = dag.task(triggerDagRun({ dagId: "downstream_etl",
waitForCompletion: true }), {
- taskId: "trigger_downstream",
- })();
+An argument filled from another task, as in ``transform(extract(), "uk")``,
arrives as that task's value. An
+upstream task that produced no value fails the task, naming the argument and
the task; one that returned
+``None`` delivers ``null``.
- trigger.after(loaded);
+A Python ``int`` larger than a JavaScript number holds exactly (beyond
±9007199254740991) is refused, so pass
+such a value as a string.
-The task has no handler to take an id from, so ``taskId`` is required, and the
factory takes no
-inputs. The trailing spec carries the task's other options, as for any task.
+Renaming an argument
+~~~~~~~~~~~~~~~~~~~~
-The task runs in the TypeScript runtime, like the Dag's other tasks, and
inherits the Dag's queue. It
-behaves as ``TriggerDagRunOperator`` does. ``waitForCompletion`` polls the run
every ``pokeInterval``
-seconds until it reaches one of ``allowedStates`` or ``failedStates``. With
``deferrable`` as well, the
-task defers to ``DagStateTrigger`` instead of holding a worker slot, and
resumes in this runtime when
-the run finishes. ``DagStateTrigger`` runs in the Python triggerer, so the
triggerer needs the standard
-provider installed.
+``withArgNames`` maps a handler's name to a different Python name, for when
the two cannot match on their own:
+a clearer word than the Dag chose, or a TypeScript reserved word such as
``enum``. The map comes first, the
+handler second:
-The task pushes the ``trigger_run_id`` XCom, and the task's "Triggered DAG"
link opens the run it
-started.
+.. code-block:: typescript
+
+ interface ReportArgs {
+ summary: Summary;
+ label: string; // the Python call passes this as `run_label`
+ }
-The task follows these config options, with Python's fallback when one is
unset:
+ const report = withArgNames({ label: "run_label" }, async ({ summary,
label }: ReportArgs) => {
+ // ...
+ });
-- ``[api] base_url`` is the base of the "Triggered DAG" link. It falls back to
``/``.
-- ``[operators] default_deferrable`` is the default of ``deferrable``. It
falls back to ``false``.
-- ``[triggerer] queues_enabled`` decides whether the deferred trigger gets the
task's queue. It falls
- back to ``false``, so the trigger gets none.
+ bundle.register(new TaskHandler("etl", "report", report));
-Some of what ``TriggerDagRunOperator`` does is not offered:
+Every name the map does not mention still matches as usual, so
``withArgNames`` should be rare. The map's keys
+are checked against the handler's argument type, so ``{ labl: "run_label" }``
is a compile error.
-- ``logical_date`` and ``run_after``. The triggered run's logical date is the
time the task triggers
- it, as in Python when neither is set.
-- Jinja. Values are sent as written, so ``{{ ds }}`` in ``conf`` reaches the
triggered run as that
- literal string.
-- OpenLineage parent injection (``openlineage_inject_parent_info``). The
runtime does not add the
- parent task's OpenLineage details to ``conf``.
+.. note::
-``new Dag`` and ``dag.task`` both take a trailing spec of Airflow options:
-``{ schedule: "@daily", tags: ["etl"] }`` for the Dag, ``{ retries: 2,
retryDelay: 30 }`` for a task.
+ A dependency drawn with ``>>`` in the Python Dag orders the tasks and passes
nothing. To read a value the
+ call did not pass, use ``getClient().getXCom``.
Writing tasks
-------------
-A task handler takes no SDK-supplied argument. Two getters, valid for as long
as the handler runs, supply what it needs:
+A handler is an ordinary, usually ``async``, function. It works the same in
both ways of using the SDK.
+
+Two functions give it what it needs while it runs:
.. list-table::
:header-rows: 1
:widths: 20 80
- * - Getter
- - Value
+ * - Function
+ - Returns
* - ``getContext()``
- - The task's execution context (a ``TaskContext``): ``dagId``, ``taskId``
(including any TaskGroup
- prefix), ``runId``, ``tryNumber``, ``mapIndex`` (``-1`` for an unmapped
task), and ``signal``, an
- ``AbortSignal`` that fires when Airflow terminates the task. Pass
``signal`` to ``fetch()``, timers,
- or other APIs that accept an ``AbortSignal`` for cooperative
cancellation.
+ - 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.
* - ``getClient()``
- - A ``TaskClient`` for Airflow Variables, Connections, and XCom.
-
-Both read a store the runtime installs around the handler call,
-which follows the handler across every ``await`` and into every promise it
creates,
-so a helper several frames deep reads them without being passed anything. Both
throw outside a handler.
+ - A ``TaskClient`` for Airflow Variables, Connections and XComs.
-Work that outlives the handler is the one gap.
-A promise the handler never awaits still resolves, but it runs after Airflow
has been told the task's terminal state,
-so await everything a handler starts.
+Both work anywhere inside the handler's call, including helper functions and
code after an ``await``, and
+throw outside it. Await everything the handler starts: work still running
after the handler returns runs
+after Airflow has recorded the task's result.
-A non-``undefined`` return value becomes the task's ``return_value`` XCom,
matching Python ``@task``
-behavior. An uncaught exception (or rejected promise) marks the task instance
failed in Airflow, triggering
-retries if configured on the stub.
+A value the handler returns becomes the task's ``return_value`` XCom, as with
Python's ``@task``, and is what a
+downstream task receives. An uncaught exception or a rejected promise fails
the task, and retries it if the task
+has retries.
-The ``TaskClient`` surface
-~~~~~~~~~~~~~~~~~~~~~~~~~~
+The ``TaskClient``
+~~~~~~~~~~~~~~~~~~
-* ``getVariable(key)`` returns the Variable as a string, or ``null`` when it
is missing;
- ``getVariableOrThrow(key)`` throws ``VariableNotFoundError`` instead,
matching Python ``Variable.get``
- with no default.
+* ``getVariable(key)`` returns the Variable as a string, or ``null`` when it
does not exist.
+ ``getVariableOrThrow(key)`` throws ``VariableNotFoundError`` instead, like
Python's ``Variable.get`` without
+ a default.
* ``setVariable(key, value, description?)`` stores a Variable, replacing any
existing value, and
- ``deleteVariable(key)`` removes one. Values are stored as strings, so
serialize structured data (for
- example with ``JSON.stringify``) before storing it.
-* ``getConnection(connId)`` returns a ``ConnectionResult`` with fields ``id``
and ``type``, plus the
- optional fields ``host``, ``schema``, ``login``, ``password``, ``port``, and
``extra`` (each may be
- missing or ``null``), or ``null`` when the connection does not exist;
- ``getConnectionOrThrow(connId)`` throws ``ConnectionNotFoundError`` instead,
matching Python
- ``BaseHook.get_connection``.
-* ``getXCom<T>({key, ...})`` reads an XCom value, or ``null`` when it is
missing. The locator fields
- (``dagId``, ``runId``, ``taskId``, ``mapIndex``) default to the current
task; pass ``taskId`` to read an
- upstream task's XCom. See :ref:`typescript-sdk/types` for how the stored
JSON maps to JavaScript types.
-* ``setXCom({key, value, ...})`` publishes an XCom value.
+ ``deleteVariable(key)`` removes one. Values are strings, so serialize
structured data, for example with
+ ``JSON.stringify``, before storing it.
+* ``getConnection(connId)`` returns a ``ConnectionResult`` with ``id`` and
``type``, plus ``host``, ``schema``,
+ ``login``, ``password``, ``port`` and ``extra``, each of which may be
missing or ``null``. It returns ``null``
+ when the connection does not exist, and ``getConnectionOrThrow(connId)``
throws ``ConnectionNotFoundError``
+ instead, like Python's ``BaseHook.get_connection``.
+* ``getXCom<T>({ key, ... })`` reads an XCom value, or ``null`` when it does
not exist. ``dagId``, ``runId``,
+ ``taskId`` and ``mapIndex`` default to the current task; pass ``taskId`` to
read an upstream task's XCom.
+ See :ref:`typescript-sdk/types` for how values map to JavaScript types.
+* ``setXCom({ key, value, ... })`` publishes an XCom value.
.. note::
- A value supplied by a secrets backend (for example an ``AIRFLOW_VAR_*``
environment variable) still
- takes precedence over the stored value when the Variable is read back.
Calling ``setVariable`` without
- a description clears the description the Variable had, and
``deleteVariable`` resolves even when the
- key does not exist.
+ A value from a secrets backend, such as an ``AIRFLOW_VAR_*`` environment
variable, still wins over the
+ stored value when the Variable is read back. ``setVariable`` without a
description clears the existing
+ description, and ``deleteVariable`` succeeds even when the Variable does
not exist.
Logging
-------
-Anything the task writes to stdout or stderr (``console.log``,
``console.error``) is captured by the worker
-and shown in the Airflow task log (stdout at ``INFO`` level, stderr at
``ERROR`` level). The SDK does not
-yet expose a dedicated structured-logging API.
+Anything a task writes to stdout or stderr, such as ``console.log`` or
``console.error``, appears in the
+Airflow task log: stdout at ``INFO`` level, stderr at ``ERROR``. There is no
dedicated structured-logging API
+yet.
.. _typescript-sdk/types:
XCom type mapping
-----------------
-XCom values are stored as JSON in Airflow's metadata database. The table below
shows how those JSON types
-surface as JavaScript values when read back via ``getXCom``.
+XComs are stored as JSON, so a value read with ``getXCom`` arrives as the
matching JavaScript type:
.. list-table::
:header-rows: 1
@@ -568,7 +550,7 @@ surface as JavaScript values when read back via ``getXCom``.
* - Python type
- JSON
- - JavaScript type (from ``getXCom``)
+ - JavaScript type
* - ``int``
- number (integer)
- ``number`` (see note)
@@ -593,65 +575,68 @@ surface as JavaScript values when read back via
``getXCom``.
.. note::
- JavaScript has a single ``number`` type (an IEEE 754 double), so integers
and decimals arrive as the same
- type, and integers larger than ``Number.MAX_SAFE_INTEGER`` (2\ :sup:`53` −
1) may lose precision.
+ JavaScript has a single ``number`` type, so integers and decimals arrive as
the same type, and an integer
+ larger than ``Number.MAX_SAFE_INTEGER`` (2\ :sup:`53` − 1) may lose
precision.
.. _typescript-sdk/build:
-Building and packaging
-----------------------
-
-``airflow-ts-pack`` (shipped with the SDK) bundles the entry module and all of
its imports with esbuild into
-a single self-contained, minified ESM file, ``bundle.min.mjs``, and embeds the
manifest (the ``dag_id`` and
-``task_id`` map plus the supervisor schema version) after a leading compact
JSON ``//# airflowBundle=...``
-layout header. The layout records the byte ranges and SHA-256 digests of the
manifest and executable code,
-so there is one file to deploy, with no separate manifest or ``node_modules``.
-
-The code is minified because an integrity digest is only worth taking over an
artifact nobody is expected to
-read or edit in place. Function names are kept through minification, since a
task id defaults to its
-handler's name. The ``/*! */`` license banners of bundled dependencies are
kept.
-
-Because the shipped code is not the code anyone wrote, the packer also embeds
source files verbatim, each in
-its own ``/*# airflowSource:<path> ... #*/`` block comment verified by its own
digest: the entry module, and
-each native Dag's file (the file its ``new Dag(...)`` constructor ran in). The
``dag_source_paths`` field in
-the manifest maps each ``dag_id`` to its file. Files that only supply
utilities or types are not embedded.
-The Code view currently shows the embedded entry module for each Dag in the
bundle.
+Building a bundle
+-----------------
-``esbuild`` is an optional peer dependency: packing is build-time only, so the
runtime install of
-``apache-airflow-ts-sdk`` skips it, and it must be installed separately before
running ``airflow-ts-pack``.
+``airflow-ts-pack``, shipped with the SDK, builds your entry module and
everything it imports into a single
+file, ``bundle.min.mjs``. That file is all you deploy: it needs no
``node_modules`` and no separate metadata
+file, and Airflow refuses to run a bundle whose content was changed after it
was built.
.. code-block:: bash
- npm install --save-dev esbuild
npx airflow-ts-pack src/main.ts --outdir dist
-Use ``--outdir <dir>`` to choose the output directory (default ``dist``),
``--outfile <path>`` to name the
-artifact exactly, which helps when one Dag bundle holds several bundles, and
``--source <name>`` to set
-the source name displayed in the Airflow UI (default: the entry file's
basename). ``--outdir`` and
-``--outfile`` are mutually exclusive, and an ``--outfile`` name must end in
``.min.mjs`` so the coordinator
-can find it.
+* ``--outdir <dir>`` sets the output directory (default ``dist``).
+* ``--outfile <path>`` names the output file exactly, which helps when one
directory holds several bundles.
+ The name must end in ``.min.mjs``.
+
+``--outdir`` and ``--outfile`` cannot be used together. ``esbuild`` must be
installed before you build; the
+runtime install of ``apache-airflow-ts-sdk`` does not need it.
+
+The bundle's code is minified, with license banners of bundled dependencies
kept. The Airflow UI's **Code**
+view shows the file each Dag is declared in, as you wrote it.
Deploying
-~~~~~~~~~
+---------
+
+Where the bundle goes depends on what it contains.
+
+Dag definition
+~~~~~~~~~~~~~~
+
+Put the bundle in a Dag bundle, such as the default ``dags-folder`` bundle.
The Dag processor runs each
+``*.min.mjs`` bundle it finds with ``node`` to read its Dags, as it reads any
other Dag file, so the Dag processor
+needs Node.js and the same ``[sdk]`` configuration as the workers:
+
+.. code-block:: ini
-Copy or mount the bundle into the Dag bundle named by the coordinator's
``task_handler_bundle_name``.
-For a task of a Python stub Dag,
-:class:`~airflow.sdk.coordinators.node.NodeCoordinator` searches that Dag
bundle recursively and launches the
-first integrity-verified ``*.min.mjs`` bundle whose metadata declares the task
instance's Dag. The artifact's
-name does not matter beyond that suffix, so one Dag bundle can hold several
bundles and a Dag is routed to
-whichever declares it. If multiple bundles declare the same Dag, the first in
sorted path order wins.
-A task of a native TypeScript Dag runs the bundle its Dag was parsed from, see
-:ref:`typescript-sdk/native-parsing`.
+ [sdk]
+ coordinators = {"ts": {"classpath":
"airflow.sdk.coordinators.node.NodeCoordinator"}}
+ queue_to_coordinator = {"typescript": "ts"}
-.. _typescript-sdk/native-parsing:
+With more than one ``NodeCoordinator``, map each Dag bundle that holds
``*.min.mjs`` bundles to one of them in
+``[sdk] dag_bundle_to_coordinator``, such as ``{"dags-folder": "ts"}``, or its
bundles fail to parse.
-Parsing native Dags
--------------------
+Their tasks run from the same Dag bundle, at the version their Dag run was
created with, so a Dag declared in
+TypeScript needs no coordinator option to find its bundle.
-A Dag declared in TypeScript ships in its packed bundle, so the bundle goes
into a Dag bundle,
-next to any Python Dag files.
-To have Airflow parse it, configure a
:class:`~airflow.sdk.coordinators.node.NodeCoordinator`.
-The Dag processor runs ``node`` on the bundle to list its Dags, so it needs
Node.js, as the workers do:
+A bundle that fails its integrity check, or whose Dags cannot be read or have
a cycle, shows up as an import
+error, and the other Dags keep working. Do not declare the same ``dag_id`` in
a Python file of that Dag bundle
+too: the two overwrite each other on every parse.
+
+Python Dag with stub TaskHandler
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+By default, the coordinator loads the bundle that implements a stub task from
the stub task's own Dag bundle,
+at the version its Dag run was created with, so the bundle goes next to the
Python Dag that declares the stub.
+
+To keep such bundles in a Dag bundle of their own, name it with
``task_handler_bundle_name``. A task then uses
+the version that bundle is on when the task starts:
.. code-block:: ini
@@ -659,55 +644,35 @@ The Dag processor runs ``node`` on the bundle to list its
Dags, so it needs Node
coordinators = {
"ts": {
"classpath": "airflow.sdk.coordinators.node.NodeCoordinator",
- "kwargs": {"node_executable": "/usr/local/bin/node"}
+ "kwargs": {"task_handler_bundle_name": "ts-handlers"}
}
}
queue_to_coordinator = {"typescript": "ts"}
-Once a ``NodeCoordinator`` is configured, the Dag processor parses the packed
bundles of every Dag bundle,
-so it needs this ``[sdk]`` configuration and Node.js. With one
``NodeCoordinator``, it parses them all. With several,
-map each Dag bundle that holds native TypeScript Dags to one of them in
``[sdk] dag_bundle_to_coordinator``.
-Each packed bundle of a Dag bundle that has no entry, or whose entry names no
``NodeCoordinator``,
-fails to parse with an import error:
-
-.. code-block:: ini
-
- [sdk]
- dag_bundle_to_coordinator = {"dags-folder": "ts"}
-
-A Dag bundle that holds only the bundles that Python stub Dags' tasks run
should list ``*`` in its ``.airflowignore``.
-Otherwise, with several Node coordinators, its bundles fail to parse.
+File names do not matter beyond the ``.min.mjs`` suffix, so one Dag bundle can
hold several bundles.
-A task of a native Dag runs the bundle file its Dag was parsed from, on the
``NodeCoordinator`` its queue routes to,
-which need not be the one that parsed it. For example, a queue can route to a
coordinator that uses another Node.js.
-A task whose queue routes to another kind of coordinator fails without
retries. A task whose bundle is missing,
-or that the coordinator cannot run (for example, because the bundle fails its
integrity check),
-fails and retries while it has retries left.
-A task of a Python stub Dag runs the first bundle, in sorted path order, that
declares its Dag.
+The Dag processor also reads a bundle that only implements stub tasks, which
starts ``node`` once per parse
+and finds no Dags. List such bundles in ``.airflowignore`` to skip that; the
coordinator still finds them.
-The Dag processor parses only files that end in ``.min.mjs`` and start with
the header ``airflow-ts-pack`` writes.
-A bundle that fails its integrity check, or whose Dags cannot be serialized or
fail validation,
-such as a cycle drawn with ``before`` and ``after``, is reported as an import
error.
+Where the configuration is needed
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-Do not declare a Dag in TypeScript that a Python file in the same bundle also
defines. When you move a Dag
-such as the Quick start's ``typescript_example`` to ``new Dag(...)``, remove
its Python stub, otherwise the
-two files overwrite each other's Dag on every parse.
+The coordinator runs where tasks run, so the ``[sdk]`` configuration and
Node.js are needed on the workers.
+With ``CeleryExecutor``, set them on the Celery workers. With
``LocalExecutor``, tasks run in the scheduler's
+process, so set them there. The Dag processor needs them to read Dags declared
in TypeScript. If it has the
+``[sdk]`` configuration, it needs Node.js even without such Dags, since it
runs every ``*.min.mjs`` bundle it
+finds. The API server never needs them.
-The Code view currently shows the bundle's entry module for each of its Dags,
as ``airflow-ts-pack``
-embeds it (see :ref:`typescript-sdk/build`). If the source cannot be read from
the bundle, the view shows a
-short notice instead.
-
-A bundle that only registers ``TaskHandler`` objects is parsed too,
-since its metadata does not say whether it declares Dags. Each parse then
launches ``node`` once and finds no Dags.
-To avoid that cost, list such bundles in ``.airflowignore``; the coordinator
still finds them to run tasks.
+There is no separate Node.js service to run: the worker starts the bundle with
``node`` once per task
+instance.
.. _typescript-sdk/coordinator-config:
:class:`~airflow.sdk.coordinators.node.NodeCoordinator` configuration
---------------------------------------------------------------------
-All ``kwargs`` in the ``coordinators`` config entry are passed to the
-:class:`~airflow.sdk.coordinators.node.NodeCoordinator` constructor:
+The ``kwargs`` of a ``coordinators`` entry are passed to
+:class:`~airflow.sdk.coordinators.node.NodeCoordinator`, and all of them are
optional:
.. list-table::
:header-rows: 1
@@ -717,47 +682,34 @@ All ``kwargs`` in the ``coordinators`` config entry are
passed to the
- Default
- Description
* - ``task_handler_bundle_name``
- - *(task's own Dag bundle)*
- - Name of the Dag bundle searched recursively for an integrity-verified
``*.min.mjs`` bundle that
- declares the requested Dag. It is used only by mixed-language Dags, to
locate the task handlers for
- the ``@task.stub`` tasks of a Python Dag; Dags defined natively in a
language SDK do not use it. It
- must be registered in ``[dag_processor] dag_bundle_config_list``. It is
checked when the ``[sdk]``
- configuration is loaded, so a typo fails there rather than on the first
task.
+ - *(unset)*
+ - The Dag bundle to load the bundles that implement stub tasks from. When
unset, a stub task loads its
+ bundle from its own Dag bundle. It must be registered in
``[dag_processor] dag_bundle_config_list``.
* - ``node_executable``
- ``"node"``
- - Path to the ``node`` binary. Defaults to ``node`` on ``$PATH``.
+ - Path to the ``node`` binary.
* - ``task_startup_timeout``
- ``10.0``
- - Seconds to wait for the Node.js subprocess to connect after launch.
Increase this if your bundle
- startup is slow (e.g. on constrained hardware).
-
-.. note::
+ - Seconds to wait for a task's bundle to start. Increase it if your
bundle starts slowly, for example on
+ constrained hardware.
- **Locating the bundle.** The packed bundles for the ``@task.stub`` tasks of
a Python Dag are read from a
- Dag bundle, so they are delivered, refreshed and versioned by the same
machinery as your Dags.
+``task_handler_bundle_name`` only concerns stub tasks: a Dag declared in
TypeScript always runs from
+the Dag bundle it was read from.
- * The expected layout is a separate Dag bundle for the packed bundles, named
by
- ``task_handler_bundle_name``, rather than the Dag bundle that holds your
``.py`` files. The task uses
- the version that Dag bundle is on when it starts, pinned for the whole
task.
- * If ``task_handler_bundle_name`` is unset, the bundle is read from the
**task's own** Dag bundle, pinned
- to the version the run was created with.
+``queue_to_coordinator`` maps a queue to a coordinator entry, so every task on
that queue runs on Node.js.
- A task of a native TypeScript Dag ignores ``task_handler_bundle_name``:
- it runs the bundle of its Dag from the Dag's own bundle, at the version the
run was created with.
- See :ref:`typescript-sdk/native-parsing`.
+.. _typescript-sdk/limitations:
Limitations
-----------
-* **A Python stub Dag is still required.** The Execution API does not yet
carry Dag structure for non-Python
- languages, so task names and dependencies are declared in Python with
- :func:`@task.stub <airflow.sdk.task.stub>`.
-* **Cluster policies do not apply to a native Dag.** ``dag_policy`` and
``task_policy`` are not run on
- it, so they cannot change or reject it.
-* **Some CLI commands do not take a native Dag.** ``airflow dags test``,
``tasks test`` and ``tasks render`` refuse it,
- and ``airflow dags reserialize`` does not store the Dags of ``*.min.mjs``
bundles,
- which only the Dag processor stores.
- ``airflow tasks list`` lists its tasks by running the bundle with ``node``,
so it needs Node.js.
-* **Beta status.** The SDK API may change in incompatible ways between
releases.
-* **One Node.js subprocess per task instance.** Tasks that need to share
in-process state between instances
- should use XCom or an external store instead.
+* **A Dag declared in TypeScript cannot express everything yet.** Assets,
custom timetables, dynamic task
+ mapping and setup and teardown tasks have no TypeScript form. Declare such a
Dag in Python, and implement
+ its tasks in TypeScript with :func:`@task.stub <airflow.sdk.task.stub>`.
+* **Cluster policies do not apply to Dags declared in TypeScript.**
``dag_policy`` and ``task_policy`` do not
+ run when the Dag processor reads them.
+* **Some CLI commands do not take a Dag declared in TypeScript.** ``airflow
dags test``, ``tasks test`` and
+ ``tasks render`` refuse it, and ``airflow dags reserialize`` skips it.
+* **Beta.** The API may change in incompatible ways between releases.
+* **One Node.js process per task instance.** Tasks cannot share in-process
state; use XComs or an external
+ store.
diff --git a/ts-sdk/README.md b/ts-sdk/README.md
index b105f768c6a..465f0ec5597 100644
--- a/ts-sdk/README.md
+++ b/ts-sdk/README.md
@@ -19,53 +19,55 @@
# Airflow TypeScript SDK
-Public TypeScript interfaces for writing Apache Airflow task handlers.
+Write Apache Airflow Dags and tasks in TypeScript.
**Status:** 0.1.0-beta1 · API may change · Node 22+ · ESM-only
-This package defines the user-facing task handler contract and the coordinator
-runtime used to execute registered TypeScript handlers from Airflow.
+There are two ways to use it:
+
+- **Dag definition**: declare the whole Dag in TypeScript, with its schedule,
tasks, options and dependencies.
+- **Python Dag with stub TaskHandler**: a TypeScript `TaskHandler` implements
each stub task of a Python Dag.
+
+Both use the same task API, and one bundle can serve both.
## Installation
```bash
npm install [email protected]
+npm install --save-dev esbuild
```
-## Task Handlers
+## Declaring a Dag
```ts
-import { Bundle, getClient, getContext, TaskHandler } from
"apache-airflow-ts-sdk";
+import { Bundle, Dag, getClient } from "apache-airflow-ts-sdk";
-export async function sayHello() {
- const greeting = await getClient().getVariable("greeting");
- return { message: `Hello from ${getContext().taskId}: ${greeting}` };
-}
+const dag = new Dag("sales_pipeline", { schedule: "@daily", queue:
"typescript" });
-const bundle = new Bundle();
-bundle.register(new TaskHandler("example_dag", "say_hello", sayHello));
-await bundle.serve();
-```
+const extract = dag.task("extract", async (): Promise<number> => {
+ return Number((await getClient().getVariable("daily_row_count")) ?? "0");
+});
+const transform = dag.task(
+ "transform",
+ async ({ rows, region }: { rows: number; region: string }) => ({ rows,
region }),
+);
-A handler is a plain function. `getContext()` and `getClient()` reach the
runtime from inside the call,
-so a handler takes no SDK-supplied parameter.
+transform({ rows: extract(), region: "us" });
-`new TaskHandler(dagId, taskId, handler)` binds the function to the
Python-owned task it implements.
-The Dag is declared in Python with `@task.stub`, so the TypeScript side only
supplies the task bodies.
+await new Bundle(dag).serve();
+```
-Non-`undefined` return values are pushed to XCom under the `"return_value"`
-key by the active runtime, matching Python `@task` behavior.
+A handler takes one object of named arguments, and calling a task names its
inputs. An input is either the
+reference another task's call returned, which makes this task wait for it and
receive its value, or a literal.
-## Coordinator Usage
+A Dag also has order-only dependencies (`before` and `after`), task groups,
branching with `dag.if` and
+`dag.switch`, and `triggerDagRun`. See the
+[guide](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/typescript.html)
+for all of them.
-Airflow runs TypeScript task bundles through the Python-side
-`airflow.sdk.coordinators.node.NodeCoordinator`. Declaring Airflow Dags in
-TypeScript is not supported yet; the Dag is still declared in Python. The
-intended authoring shape matches the other non-Python SDKs: a Python Dag
-declares the scheduling shape with stub tasks, and the TypeScript module
-registers handlers with matching task IDs.
+## Implementing stub tasks of a Python Dag
-Python Dag:
+The Python Dag declares the tasks and their dependencies:
```python
from airflow.sdk import dag, task
@@ -85,107 +87,36 @@ def sales_pipeline():
sales_pipeline()
```
-Airflow coordinator config:
-
-```ini
-[dag_processor]
-dag_bundle_config_list = [
- {"name": "dags-folder", "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
- {
- "name": "ts-task-handlers",
- "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
- "kwargs": {"path": "/opt/airflow/ts-bundles"}
- }
- ]
-
-[sdk]
-coordinators = {
- "ts": {
- "classpath": "airflow.sdk.coordinators.node.NodeCoordinator",
- "kwargs": {"task_handler_bundle_name": "ts-task-handlers"}
- }
-}
-queue_to_coordinator = {"typescript": "ts"}
-```
-
-The Dag bundle named by `task_handler_bundle_name` (the task's own Dag bundle
when unset) is searched
-recursively for `*.min.mjs` bundles built with `airflow-ts-pack` (see [Packing
bundles](#packing-bundles)),
-which embeds the Airflow metadata in the bundle itself.
`task_handler_bundle_name` is used only by
-mixed-language Dags, to locate the task handlers for the `@task.stub` tasks of
a Python Dag; Dags defined
-natively in a language SDK do not use it.
-
-This config, the packed bundles and Node.js 22 or later must be present
wherever tasks execute and on the
-Dag processor, which checks the stub tasks of each Python Dag against the task
handlers the bundles
-register. The API server does not need them.
-
-TypeScript entrypoint:
+TypeScript implements them, with a `TaskHandler` per stub task:
```ts
import { Bundle, getClient, TaskHandler } from "apache-airflow-ts-sdk";
export async function extract() {
- const client = getClient();
- const connection = await client.getConnection("sales_db");
- const rowCount = Number((await client.getVariable("daily_row_count")) ??
"0");
-
- return {
- connectionId: connection?.id ?? null,
- rowCount,
- };
+ const rowCount = Number((await getClient().getVariable("daily_row_count"))
?? "0");
+ return { rowCount };
}
-export async function transform() {
- const extracted = await getClient().getXCom<{ rowCount: number }>({
- key: "return_value",
- taskId: "extract",
- });
-
- return {
- transformedRows: extracted?.rowCount ?? 0,
- };
+export async function transform({ extracted }: { extracted: { rowCount: number
} }) {
+ return { transformedRows: extracted.rowCount };
}
-const bundle = new Bundle();
-bundle.register(
- new TaskHandler("sales_pipeline", "extract", extract),
- new TaskHandler("sales_pipeline", "transform", transform),
-);
-await bundle.serve();
-```
-
-The Python stub defines the Dag dependency graph. The TypeScript handler does
-the work and uses `TaskClient` for task-time Airflow data access. The handler
-function is the reusable task implementation; a `TaskHandler` binds it to a
-Python stub task identity, a `Bundle` holds what this bundle process provides,
-and `bundle.serve()` serves it to Airflow.
-
-`bundle.serve()` is the entrypoint: a task left unregistered is not part of
the bundle,
-and one bundle can provide for several `TaskHandler`s.
-Registering holds no sockets and starts nothing, so a unit test can build a
bundle
-and dispatch through `bundle.getTaskHandler(dagId, taskId)` without any
runtime involved.
-
-Dispatch keys on the `(dagId, taskId)` pair, so the same `taskId` under two
Dags is two different handlers:
-
-```ts
-import { Bundle, TaskHandler } from "apache-airflow-ts-sdk";
-import { chargeCustomer } from "./billing/tasks";
-import { extract } from "./sales/tasks";
-
await new Bundle(
new TaskHandler("sales_pipeline", "extract", extract),
- new TaskHandler("billing_pipeline", "extract", chargeCustomer),
+ new TaskHandler("sales_pipeline", "transform", transform),
).serve();
```
-Register `TaskHandler` and `Dag` values with the `register` method, or pass
them to the `Bundle` constructor.
+A `TaskHandler` names the `dag_id` and `task_id` it implements, so the same
`task_id` under two Dags is two
+different handlers. Nothing connects to Airflow until `bundle.serve()`, so a
unit test can build a bundle and
+call a handler through `bundle.getTaskHandler(dagId, taskId)`.
-## TaskFlow arguments
+### TaskFlow arguments
-A Python Dag that calls a stub task TaskFlow-style passes those arguments
straight to the handler,
-which destructures them by name:
+A Python Dag that calls a stub task TaskFlow-style passes those arguments 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): ...
@@ -194,133 +125,55 @@ 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.
-
-### Upstream outputs
-
-An argument the Python call fills from another task arrives as that task's
value, not as a reference to it.
-Feeding the `transform` above from an upstream task adds one argument on each
side:
-
-```python
-# the Python Dag
-@task
-def extract() -> int: ...
-
-
[email protected](queue="typescript")
-def transform(rows: int, region_code: str, threshold: float, dry_run: bool =
False): ...
-
-
-transform(extract(), "uk", 0.75)
-```
-
-```ts
-interface TransformArgs {
- rows: number;
- regionCode: string;
- threshold: number;
- dryRun: boolean;
-}
-
-export async function transform({ rows, regionCode, threshold, dryRun }:
TransformArgs) {
- // `rows` is the number extract() returned.
-}
-```
-
-An upstream that pushed no output fails the task, naming both the argument and
the task it came from.
-An upstream that pushed `null` binds `null`.
-
-Being upstream is not the same as being passed. An XCom dependency declared
with `>>` defines task order only,
-so a value the call did not pass is read explicitly:
-
-```ts
-const rows = await getClient().getXCom<number>({ key: "return_value", taskId:
"extract" });
-```
-
-A Python `int` beyond the ±9007199254740991 a JavaScript number holds exactly
is refused rather than bound,
-so carry such a value across the boundary as a string.
-
-### Explicit renames
+Names match ignoring case and underscores, so `region_code` reaches
`regionCode` with nothing declared on
+either side. An argument filled from another task, as in `transform(extract(),
"uk")`, arrives as that task's
+value. A dependency drawn with `>>` orders the tasks and passes nothing; read
such a value with
+`getClient().getXCom`.
-`withArgNames` states a binding folding cannot reach, for a name the Python
side never used:
-a clearer word than the Dag chose, or a TypeScript reserved word like `enum`.
Mapping first, handler second:
+`withArgNames` maps a handler's name to a different Python name, when the two
cannot match on their own:
```ts
-interface ReportArgs {
- summary: Summary;
- label: string; // Python calls this `run_label`
-}
-
const report = withArgNames({ label: "run_label" }, async ({ summary, label }:
ReportArgs) => {
- // `label` is the call's `run_label`; `summary` folded as usual.
+ // `label` is the call's `run_label`.
});
-
-bundle.register(new TaskHandler("etl", "report", report));
```
-An entry beats folding, and everything the map does not mention still folds,
-so `withArgNames` should be rare in a real Dag.
-The map's keys are checked against the handler's own parameter type,
-so `{ labl: "run_label" }` is a compile error naming the right key.
-Its values are Python names, which `tsc` cannot see and does not check.
-
-`Dag` is another interface, for a Dag declared natively in TypeScript, and is
still a work in progress.
+## Writing tasks
-Airflow launches the bundled entrypoint with `--comm=host:port` and
-`--logs=host:port`. `bundle.serve()` connects to those sockets, receives the
-task startup message, finds the registered handler for the Dag/task pair, and
-reports the terminal task state back to Airflow.
+A handler is a plain, usually `async`, function. `getContext()` and
`getClient()` give it the task's context and
+Airflow access while it runs, so it takes no SDK-supplied argument. A value it
returns becomes the task's
+`return_value` XCom, and an uncaught error fails the task.
-See [`example/`](https://github.com/apache/airflow/tree/main/ts-sdk/example)
for
-a coordinator-runtime example that packs a bundle with `airflow-ts-pack` and
-uses a Python stub Dag.
+## Building and deploying
-## Packing bundles
-
-`airflow-ts-pack` produces a single self-contained bundle in one command.
-Packing is build-time only, so `esbuild` is an optional peer dependency the
-runtime install skips:
+`airflow-ts-pack` builds the entry module and everything it imports into one
file, `dist/bundle.min.mjs`:
```bash
-npm install --save-dev esbuild
-airflow-ts-pack src/main.ts --outdir dist
+npx airflow-ts-pack src/main.ts --outdir dist
```
-It bundles the entrypoint into a minified `dist/bundle.min.mjs` with esbuild,
then runs that bundle with
-`--airflow-metadata` so it reports its own registered Dag/task pairs and
supervisor schema version. The manifest is
-embedded as a compact JSON `//# airflowMetadata=...` comment after a leading
compact JSON `//# airflowBundle=...`
-layout descriptor, and the entry module is embedded verbatim in a `/*#
airflowSource ... #*/` block comment so
-Airflow can show the source a bundle was authored from, which its minified
code no longer is. The CLI records the
-integrity metadata for all three regions in that descriptor, so a coordinator
that is handed a bundle whose content
-was replaced fails loudly instead of running it. The result is one deployable
file with no hand-written metadata
-sidecar.
+- `--outdir <dir>`: output directory (default `dist`)
+- `--outfile <path>`: exact output path, whose name must end in `.min.mjs`
-Pass `--outfile <path>` instead of `--outdir` to name the artifact yourself,
so one bundle directory can hold several
-bundles. The name must still end in `.min.mjs`, which is how `NodeCoordinator`
finds bundles.
+A bundle that declares Dags goes into a Dag bundle, such as the default
`dags-folder` bundle. Route its queue to
+the Node.js coordinator:
-Options:
+```ini
+[sdk]
+coordinators = {"ts": {"classpath":
"airflow.sdk.coordinators.node.NodeCoordinator"}}
+queue_to_coordinator = {"typescript": "ts"}
+```
-- `--outdir <dir>`: output directory (default `dist`)
-- `--outfile <path>`: exact output path, whose name must end in `.min.mjs`
-- `--source <name>`: display name of the primary source file shown in the
Airflow UI (default: entry basename)
+A bundle that only implements stub tasks goes next to the Python Dag that
declares them, or into the Dag
+bundle the coordinator's `task_handler_bundle_name` names. See the guide for
all coordinator options.
+
+See [`example/`](https://github.com/apache/airflow/tree/main/ts-sdk/example)
for a working project that serves
+both a Dag declared in TypeScript and the tasks of two Python Dags from one
bundle.
## TaskClient
diff --git a/ts-sdk/docs/index.md b/ts-sdk/docs/index.md
index 7d60fb0d945..c3ba05b440d 100644
--- a/ts-sdk/docs/index.md
+++ b/ts-sdk/docs/index.md
@@ -19,9 +19,8 @@
# Apache Airflow TypeScript SDK
-The Apache Airflow TypeScript SDK provides the public TypeScript interfaces for
-writing Apache Airflow task handlers and the coordinator runtime that executes
-registered TypeScript handlers from an Airflow worker.
+Write Apache Airflow Dags and tasks in TypeScript. Declare a whole Dag in
TypeScript, or implement the stub
+tasks of a Python Dag with `TaskHandler`.
> **Note**
> This package is **0.1.0-beta1**: the API may change, it requires **Node
> 22+**, and
@@ -29,61 +28,50 @@ registered TypeScript handlers from an Airflow worker.
## Getting Started
-Install the beta package from npm:
+Install the beta package from npm, and `esbuild` to build the bundle:
```bash
npm install [email protected]
+npm install --save-dev esbuild
```
-Bind a handler to the Python-owned task it implements, register it on a
`Bundle`, and serve it.
-A handler is a plain function: `getContext()` returns the `TaskContext` and
`getClient()` the `TaskClient`
-for as long as it runs, so neither is a parameter.
-Any non-`undefined` return value is pushed to XCom under the `"return_value"`
key by the active runtime,
-matching Python `@task` behavior:
+Declare a Dag with `Dag`, and its tasks with `dag.task`. Calling a task names
its inputs, which is how the
+tasks depend on each other:
```ts
-import { Bundle, getClient, getContext, TaskHandler } from
"apache-airflow-ts-sdk";
+import { Bundle, Dag } from "apache-airflow-ts-sdk";
-export async function sayHello() {
- const greeting = await getClient().getVariable("greeting");
- return { message: `Hello from ${getContext().taskId}: ${greeting}` };
-}
+const dag = new Dag("ts_etl", { schedule: "@daily", queue: "typescript" });
-const bundle = new Bundle();
-bundle.register(new TaskHandler("example_dag", "say_hello", sayHello));
-await bundle.serve();
-```
+const extract = dag.task("extract", async (): Promise<number> => 42);
+const load = dag.task("load", async ({ rows }: { rows: number }) => {
+ console.log(`loading ${rows} rows`);
+});
-`register` takes any number of items, so one bundle can provide for several
`TaskHandler`s.
+load({ rows: extract() });
+
+await new Bundle(dag).serve();
+```
-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:
+To implement a stub task of a Python Dag, bind a function to its `dag_id` and
`task_id` with `TaskHandler`:
```ts
-interface TransformArgs {
- regionCode: string;
- threshold: number;
-}
+import { Bundle, getClient, getContext, TaskHandler } from
"apache-airflow-ts-sdk";
-export async function transform({ regionCode, threshold }: TransformArgs) {
- // ...
+export async function sayHello() {
+ const greeting = await getClient().getVariable("greeting");
+ return { message: `Hello from ${getContext().taskId}: ${greeting}` };
}
-```
-`withArgNames` states a binding folding cannot reach, for a name the Python
side never used.
-It should be rare, since folding covers ordinary spelling differences.
-
-`Dag` is another interface, for a Dag declared in TypeScript rather than in
Python, and is still a work in progress.
+await new Bundle(new TaskHandler("example_dag", "say_hello",
sayHello)).serve();
+```
-## Coordinators
+In both cases a handler is a plain function. `getContext()` returns the
`TaskContext` and `getClient()` the
+`TaskClient` while it runs, and a value it returns becomes the task's
`return_value` XCom.
-Airflow runs TypeScript task bundles through the Python-side `NodeCoordinator`
-(`airflow.sdk.coordinators.node.NodeCoordinator`). A Python Dag declares the
-scheduling shape with stub tasks and owns the task dependencies between them,
-and the TypeScript module registers handlers with matching task IDs. See the
-[Non-Python Task SDKs
guide](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/index.html)
-for the conceptual overview of language SDKs.
+See the
+[TypeScript SDK
guide](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/typescript.html)
+for building and deploying a bundle, and for everything a Dag declared in
TypeScript can do.
## API Reference
diff --git a/ts-sdk/example/README.md b/ts-sdk/example/README.md
index 85b5c74099f..60064cc6557 100644
--- a/ts-sdk/example/README.md
+++ b/ts-sdk/example/README.md
@@ -19,17 +19,18 @@
# TypeScript Coordinator Runtime Example
-This example shows the coordinator-mode shape for TypeScript task handlers:
-
-- `dags/typescript_example.py` and `dags/typescript_taskflow_example.py`
declare two Airflow Dags and their stub tasks.
-- `src/main.ts` and `src/taskflow.ts` register a `TaskHandler` per stub task
and start the coordinator runtime.
- One bundle provides for both Dags, and both declare a task called
`build_message`.
- A handler binds the `(dag_id, task_id)` pair, so the two are different tasks
with different bodies.
-- `dist/bundle.min.mjs` is the generated Node.js bundle that Airflow launches.
-
-The build uses the SDK's `airflow-ts-pack` tool, which bundles the entrypoint
-with esbuild and embeds the Airflow metadata generated from the bundle's
-registered tasks, producing a single deployable file.
+This example shows both ways to use the TypeScript SDK, served from one bundle:
+
+- `src/native.ts` declares the `typescript_native_example` Dag entirely in
TypeScript: its schedule, task
+ options and dependencies, including a task group, a condition, a multi-way
branch and a triggered Dag run.
+- `dags/typescript_example.py` and `dags/typescript_taskflow_example.py`
declare two Python Dags with stub
+ tasks, and `src/main.ts` and `src/taskflow.ts` implement them with a
`TaskHandler` per stub task.
+ Both Dags have a task called `build_message`; a handler names its `(dag_id,
task_id)` pair, so the two are
+ different tasks with different bodies.
+- `dist/bundle.min.mjs` is the built bundle Airflow runs.
+
+The build uses the SDK's `airflow-ts-pack` tool, which builds the entry module
and everything it imports into
+that single file.
## Build
