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 b6538c32c8b TS SDK: declare task dependencies by calling tasks (#73435)
b6538c32c8b is described below
commit b6538c32c8b1cdcb7b818e174a8b09d69ae18ad6
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Wed Sep 23 10:33:07 2026 +0800
TS SDK: declare task dependencies by calling tasks (#73435)
---
.../language-sdks/typescript.rst | 49 ++-
ts-sdk/adr/0002-native-dag-interface.md | 32 +-
ts-sdk/src/coordinator/manifest.ts | 7 +-
ts-sdk/src/coordinator/runtime.ts | 12 +-
ts-sdk/src/index.ts | 11 +-
ts-sdk/src/sdk/bundle.ts | 195 ++++-----
ts-sdk/src/sdk/dag.ts | 437 +++++++++++++++++----
ts-sdk/tests/cli/fixtures/entry.ts | 2 +-
ts-sdk/tests/cli/fixtures/noisy-entry.ts | 2 +-
ts-sdk/tests/cli/pack.test.ts | 10 +-
ts-sdk/tests/coordinator/integration.test.ts | 43 +-
ts-sdk/tests/coordinator/runtime-manifest.test.ts | 4 +-
ts-sdk/tests/public-api.test.ts | 89 ++++-
ts-sdk/tests/sdk/bundle.test.ts | 82 ++--
ts-sdk/tests/sdk/dag.test.ts | 366 ++++++++++++++---
ts-sdk/tests/sdk/task-handler.test.ts | 64 ++-
16 files changed, 1039 insertions(+), 366 deletions(-)
diff --git
a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
index b954ff9cd6c..aad721a1a4f 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -121,16 +121,12 @@ That top-level ``await`` makes the module a runnable
bundle entry point.
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.
-``register`` takes any number of task handlers 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.
+``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.
-``Dag`` is another interface, for a Dag declared in TypeScript rather than in
Python, and is still a work
-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
~~~~~~~~~~~~~~~~~~
@@ -242,6 +238,47 @@ task instance.
inside the scheduler process, so they must be present where the scheduler
can read them. The API server
and Dag processor do not need them.
+.. _typescript-sdk/native-dag:
+
+Declaring a Dag in TypeScript
+-----------------------------
+
+A ``Dag`` is declared on this side rather than in Python: its tasks and the
edges between them are
+written in TypeScript. The surface is still growing, so a Dag declared this
way is not served to
+Airflow yet.
+
+``dag.task(taskId, handler)`` returns a *factory*. Calling it places the task
in the Dag and supplies the
+handler's arguments, so the call graph is the task graph:
+
+.. code-block:: typescript
+
+ import { Dag } from "apache-airflow-ts-sdk";
+
+ const dag = new Dag("ts_etl");
+
+ const extract = dag.task("extract", async (): Promise<number> => 42);
+ const transform = dag.task("transform", async (rows: number, region:
string) => rows * 2);
+ const load = dag.task("load", async (total: number) => {});
+
+ load(transform(extract(), "us"));
+
+Arguments are passed in the order the handler declares them. A handler that
declares a single object of
+named arguments can also be called with that object, which names each input
instead of ordering it:
+
+.. code-block:: typescript
+
+ const store = dag.task("store", async ({ total }: { total: number }) =>
{});
+
+ store({ total: extract() });
+
+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.
+
+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.
+
+``new Dag`` and ``dag.task`` also take a trailing ``spec`` object that is not
used yet; do not set it.
+
Writing tasks
-------------
diff --git a/ts-sdk/adr/0002-native-dag-interface.md
b/ts-sdk/adr/0002-native-dag-interface.md
index 2c2a1a3df1b..eca1f5e2ea2 100644
--- a/ts-sdk/adr/0002-native-dag-interface.md
+++ b/ts-sdk/adr/0002-native-dag-interface.md
@@ -28,11 +28,15 @@ Proposed. Revised after the review on #72047.
1. **`dag.task(handler)` returns a factory, and the task id is optional.**
With no id the task takes
the handler's function name (`dag.task(extract)` → task `"extract"`);
`dag.task(taskId, handler)`
sets it explicitly, which an anonymous handler must do. Calling the factory
both places the task in
- the Dag and supplies its arguments by name — the shape Python TaskFlow uses
for
+ the Dag and supplies its arguments, in the order the handler declares them
+ (`load(transform(extract(), "us"))`). A handler that declares a single
object of named arguments can
+ also be called with that object — the shape Python TaskFlow uses for
`load(transformed=transform(...))`.
2. **The call graph is the task graph.** `tsc` checks every wired key against
the handler's own
parameter type, and a `TaskRef` exists only once its producing call has
returned, so a cycle
- through arguments is unrepresentable rather than rejected by a validator.
+ through arguments is unrepresentable rather than rejected by a validator. A
reference passed by
+ position is checked against the argument's own type, which is what tells
the two call shapes apart
+ when a handler declares a single argument.
3. **Every task is called exactly once.** An uncalled task fails when the Dag
is read, so none can be
silently left out of the graph.
4. **`before` and `after` draw order-only edges** — the TypeScript pair for
`>>` and `<<`, both
@@ -145,9 +149,9 @@ convention.
by design. Native declaration is what fills them, generated from the
serialized-Dag JSON schema the
way `src/generated/supervisor.ts` is. This ADR does not choose those fields;
it fixes where an
author writes them.
-- `TaskOptions` collapses into `TaskSpec`. The shipped third argument to
`dag.task` is
- `{ inputs, spec }`; with wiring moved to the factory call, `inputs` is no
longer an option and the
- third argument is the spec itself.
+- `TaskOptions` carries the spec and the handler's positional argument names,
which the packer fills in
+ from the parameter list so the Dag names each argument as its handler does.
With wiring moved to the
+ factory call, `inputs` is no longer an option.
- `TaskHandlerArgs` is removed from the public API, `DagRegistry` becomes
`Bundle`, and
`serveDags(registry)` becomes `bundle.serve()`, which breaks
0.1.0-beta1 authors; see [ADR-0001](0001-mixed-lang-dag-interface.md) for
the shipped call sites
@@ -155,11 +159,10 @@ convention.
## Alternatives
-- **Positional wiring** (`load(transform(extract()))`), which becomes
expressible once data no longer
- shares an object with `ctx`/`client`, since a handler can then take its
arguments positionally and
- `Parameters<typeof handler>` is a real tuple. Rejected: it removes the key
names from every call
- site, and those names are what keeps flat, one-statement-per-task wiring
readable at twenty tasks.
- A handler may still take several positional arguments; only the *wiring*
stays named.
+- **Named-only wiring**, rejected in the review on #73435: naming every input
reads well at twenty
+ tasks but forces an object around a single argument, and positional calls
are what TypeScript
+ authors write. Both are offered, and the handler's own parameter list
decides which one a task can
+ use.
- **Injected `ctx`/`client` arguments**, mimicking the Python signature.
Rejected, per the above and
because feeling native to TypeScript matters more than matching Python's
parameter list.
@@ -175,9 +178,12 @@ convention.
the same object, which forced every typed handler to declare `TArgs &
TaskHandlerArgs` and left the
top-level argument namespace open to collisions with an author's own
parameter names. Getters close
both.
-- **The spec argument already has its slot.** `dag.task(taskId, handler,
options)` reads
- `{ inputs = {}, spec = {} }` and runs `validateEmptySpec` on the spec today
- (`ts-sdk/src/sdk/dag.ts`), so task fields land on a path that exists rather
than a new one.
+- **The spec argument already has its slot.** `dag.task(taskId, handler,
options)` reads `{ spec = {} }`
+ and runs `validateEmptySpec` on it (`ts-sdk/src/sdk/dag.ts`), so task fields
land on a path that
+ exists rather than a new one.
+- **A positional argument binds by order, and its name is a label.** The
serialized Dag names each
+ argument, so the packer reads the names from the handler's parameter list;
`arg0`, `arg1` and so on
+ stand in for a name it cannot see, without changing which value reaches
which argument.
- **A `TaskRef` is inert** — a handle for wiring, not a promise. Nothing in a
Dag file executes a task
body.
- **A defaulted task id is resolved at pack time, not read at runtime.**
esbuild renames function
diff --git a/ts-sdk/src/coordinator/manifest.ts
b/ts-sdk/src/coordinator/manifest.ts
index 97661257efd..523385236a3 100644
--- a/ts-sdk/src/coordinator/manifest.ts
+++ b/ts-sdk/src/coordinator/manifest.ts
@@ -18,7 +18,7 @@
*/
import { SUPERVISOR_API_VERSION } from "./protocol.js";
-import { listBundleDags, type Bundle } from "../sdk/bundle.js";
+import { bundleDagTaskIds, finalizeBundleDags, type Bundle } from
"../sdk/bundle.js";
export const AIRFLOW_METADATA_FLAG = "--airflow-metadata";
@@ -36,7 +36,10 @@ export interface BundleManifest {
export function buildBundleManifest(bundle: Bundle): BundleManifest {
const taskHandlers: BundleManifest["task_handlers"] = {};
- for (const { dagId, tasks } of listBundleDags(bundle)) {
+ // The manifest is the bundle reporting what it provides, so this is where
its
+ // Dags are finalized: a Dag missing an edge is reported here rather than
packed.
+ finalizeBundleDags(bundle);
+ for (const [dagId, tasks] of bundleDagTaskIds(bundle)) {
if (typeof dagId !== "string") {
throw new Error("Dag ID must be a string");
}
diff --git a/ts-sdk/src/coordinator/runtime.ts
b/ts-sdk/src/coordinator/runtime.ts
index deb0cb485a0..bf9764f3e66 100644
--- a/ts-sdk/src/coordinator/runtime.ts
+++ b/ts-sdk/src/coordinator/runtime.ts
@@ -55,7 +55,7 @@ import {
type StartupDetails,
} from "./protocol.js";
import { getArgNames } from "../sdk/arg-names.js";
-import { listBundleTasks, type Bundle } from "../sdk/bundle.js";
+import { bundleDagTaskIds, type Bundle } from "../sdk/bundle.js";
import { runInTaskScope, type TaskContext } from "../sdk/task.js";
import type { JsonValue } from "../sdk/client-types.js";
@@ -148,10 +148,10 @@ export async function startCoordinator(
const runtimeLogs = logs.child("runtime");
runtimeLogs.debug("Connecting log socket", { logs_addr: parsed.logsAddr });
await logs.connect(parsed.logsAddr);
- const tasks = listBundleTasks(bundle);
+ const byDag = bundleDagTaskIds(bundle);
runtimeLogs.info("Coordinator runtime started", {
- registered_tasks: tasks,
- count: tasks.length,
+ registered_tasks: Object.fromEntries(byDag),
+ count: [...byDag.values()].reduce((total, tasks) => total +
tasks.length, 0),
// Cadwyn schema version this SDK was generated against. Logged
// for operator visibility; not sent on the wire.
supervisor_api_version: SUPERVISOR_API_VERSION,
@@ -270,7 +270,7 @@ function handleParse(
// TypeScript-native Dag parsing is not yet supported.
// Respond with an empty result so the Python-stub-Dag workflow works.
logs.info("Parse-mode response (TS Dag parsing not yet supported)", {
- registered_tasks: listBundleTasks(bundle),
+ registered_tasks: Object.fromEntries(bundleDagTaskIds(bundle)),
});
const response: RuntimeDagFileParsingResult = {
type: "DagFileParsingResult",
@@ -295,7 +295,7 @@ async function handleTask(
logs.warning("No handler registered for task", {
dag_id: ti.dag_id,
task_id: ti.task_id,
- available: listBundleTasks(bundle),
+ available: Object.fromEntries(bundleDagTaskIds(bundle)),
});
// A missing handler means this bundle cannot run the task, so retrying the
// same bundle/configuration mismatch would not help.
diff --git a/ts-sdk/src/index.ts b/ts-sdk/src/index.ts
index e6d162155e3..d63bdf24186 100644
--- a/ts-sdk/src/index.ts
+++ b/ts-sdk/src/index.ts
@@ -26,7 +26,16 @@ export { ConnectionNotFoundError, VariableNotFoundError }
from "./sdk/client.js"
export { SUPERVISOR_API_VERSION } from "./coordinator/index.js";
export type { ArgNameMap } from "./sdk/arg-names.js";
export type { Registerable } from "./sdk/bundle.js";
-export type { DagSpec, TaskInputs, TaskOptions, TaskRef, TaskSpec } from
"./sdk/dag.js";
+export type {
+ DagSpec,
+ PositionalInputs,
+ TaskFactory,
+ TaskInput,
+ TaskInputs,
+ TaskOptions,
+ TaskRef,
+ TaskSpec,
+} from "./sdk/dag.js";
export type { TaskClient } from "./sdk/client.js";
export type { ConnectionResult, GetXComOpts, JsonValue, SetXComOpts } from
"./sdk/client-types.js";
export type { TaskContext, TaskFunction } from "./sdk/task.js";
diff --git a/ts-sdk/src/sdk/bundle.ts b/ts-sdk/src/sdk/bundle.ts
index ad99d0c8f34..2aad2aa03bc 100644
--- a/ts-sdk/src/sdk/bundle.ts
+++ b/ts-sdk/src/sdk/bundle.ts
@@ -20,12 +20,13 @@
// The bundle: what a TypeScript bundle process provides, and how it serves it.
import { brand, DUPLICATE_COPY_HINT, hasBrand } from "./brand.js";
-import { Dag, getDagTaskRecords, isDag, type TaskRef } from "./dag.js";
+import { Dag, finalizeDag, getDagTaskRecords, isDag } from "./dag.js";
import { getTaskHandlerFunction, isTaskHandler, TaskHandler } from
"./task-handler.js";
import type { TaskFunction } from "./task.js";
// Assigned inside Bundle's static block, as Dag does for its tasks.
-let entriesOf: (bundle: Bundle) => ReadonlyMap<string, BundleEntry>;
+let dagsOf: (bundle: Bundle) => ReadonlyMap<string, Dag>;
+let taskHandlersOf: (bundle: Bundle) => ReadonlyMap<string,
ReadonlyMap<string, TaskFunction>>;
/**
* What {@link Bundle.register} and the {@link Bundle} constructor accept: a
@@ -37,32 +38,11 @@ let entriesOf: (bundle: Bundle) => ReadonlyMap<string,
BundleEntry>;
// is assignable to, `TaskHandler<TransformArgs>` included.
export type Registerable = Dag | TaskHandler<never, unknown>;
-// What a bundle holds per dag_id. The two arms are exclusive by construction:
-// a Dag is the native case and owns its own tasks, while task handlers supply
-// bodies for a Dag that Python declares, so one dag_id is never both.
-type BundleEntry =
- | { readonly kind: "dag"; readonly dag: Dag }
- | { readonly kind: "handlers"; readonly handlers: Map<string, TaskFunction>
};
-
-function entryTaskIds(entry: BundleEntry): string[] {
- return entry.kind === "dag" ? [...entry.dag.taskIds] :
[...entry.handlers.keys()];
-}
-
/** Internal: whether `value` is a Bundle built by any copy of this package. */
export function isBundle(value: unknown): value is Bundle {
return hasBrand(value, "Bundle");
}
-/** Internal: a Dag this bundle provides for, with its task IDs, as
- * {@link listBundleDags} reports it. A task-less native Dag is included, so
- * the bundle manifest keeps it visible. */
-export interface RegisteredDag {
- /** Identifier of the registered Dag. */
- readonly dagId: string;
- /** Airflow task IDs, including any TaskGroup prefix. */
- readonly tasks: string[];
-}
-
/**
* What a bundle process provides to Airflow, and the thing that serves it.
*
@@ -87,12 +67,15 @@ export interface RegisteredDag {
* a snapshot of its tasks.
*/
export class Bundle {
- // Keyed by dag_id and insertion-ordered, so the bundle manifest lists what
- // this process provides in the order the entry point registered it.
- readonly #entries = new Map<string, BundleEntry>();
+ // Dags declared in TypeScript, keyed by dag_id in registration order.
+ #dags = new Map<string, Dag>();
+ // Handlers for the tasks a Python Dag declares, keyed by dag_id and then by
+ // task_id, both in registration order.
+ #taskHandlers = new Map<string, Map<string, TaskFunction>>();
static {
- entriesOf = (bundle) => bundle.#entries;
+ dagsOf = (bundle) => bundle.#dags;
+ taskHandlersOf = (bundle) => bundle.#taskHandlers;
}
/** Registers `items`, on the same terms as {@link register}. */
@@ -107,11 +90,13 @@ export class Bundle {
* The constructor covers the common case; this is for a bundle that
* collects what it provides across several modules. */
register(...items: Registerable[]): void {
- // Validated against what is already held *and* against this call, in full,
- // before anything is written: a call that throws registers none of its
- // items, so a bundle never half-provides what its author listed once.
- const incomingDags = new Set<string>();
- const incomingTaskHandlers = new Set<string>();
+ // Staged on copies and committed once every item is accepted, so a call
+ // that throws registers none of its items and a bundle never
+ // half-provides what its author listed in one call.
+ const dags = new Map(this.#dags);
+ const taskHandlers = new Map(
+ [...this.#taskHandlers].map(([dagId, handlers]) => [dagId, new
Map(handlers)] as const),
+ );
for (const item of items) {
// Typed as Registerable, so narrowing it would collapse to never; these
// guard callers reaching this from plain JavaScript.
@@ -119,9 +104,9 @@ export class Bundle {
// Another copy's value cannot be registered, since both kinds read
// private state keyed to this copy's class, so it is rejected by its
cause.
if (candidate instanceof Dag) {
- this.#checkDag(candidate, incomingDags);
+ stageDag(dags, taskHandlers, candidate);
} else if (candidate instanceof TaskHandler) {
- this.#checkTaskHandler(candidate, incomingDags, incomingTaskHandlers);
+ stageTaskHandler(dags, taskHandlers, candidate);
} else if (isDag(candidate)) {
throw new Error(`Dag "${(candidate as Dag).dagId}"
${DUPLICATE_COPY_HINT}`);
} else if (isTaskHandler(candidate)) {
@@ -133,61 +118,8 @@ export class Bundle {
throw new Error("only Dag and TaskHandler instances can be
registered");
}
}
- for (const item of items) {
- if (item instanceof Dag) {
- this.#entries.set(item.dagId, { kind: "dag", dag: item });
- } else {
- const existing = this.#entries.get(item.dagId);
- const handlers =
- existing?.kind === "handlers" ? existing.handlers : new Map<string,
TaskFunction>();
- handlers.set(item.taskId, getTaskHandlerFunction(item));
- if (existing === undefined) {
- this.#entries.set(item.dagId, { kind: "handlers", handlers });
- }
- }
- }
- }
-
- #checkDag(dag: Dag, incomingDags: Set<string>): void {
- if (this.#entries.get(dag.dagId)?.kind === "handlers") {
- throw new Error(
- `Dag "${dag.dagId}" already has registered task handlers; a Dag
declared in ` +
- "TypeScript owns its own tasks, so one Dag ID cannot have both",
- );
- }
- if (this.#entries.has(dag.dagId) || incomingDags.has(dag.dagId)) {
- throw new Error(`Dag "${dag.dagId}" is already registered`);
- }
- incomingDags.add(dag.dagId);
- }
-
- #checkTaskHandler(
- handler: TaskHandler,
- incomingDags: Set<string>,
- incomingTaskHandlers: Set<string>,
- ): void {
- // A native Dag attaches its tasks with dag.task(...), so a task handler
for
- // the same Dag ID would be a second, disagreeing source for its task list.
- if (this.#entries.get(handler.dagId)?.kind === "dag" ||
incomingDags.has(handler.dagId)) {
- throw new Error(
- `Dag "${handler.dagId}" is declared in TypeScript; attach its tasks
with ` +
- "dag.task(...) rather than registering task handlers for them",
- );
- }
- const entry = this.#entries.get(handler.dagId);
- // Keyed on the pair, not the task ID: one bundle serves several Dags, and
- // the same task_id under two of them is two different handlers.
- const pair = `${handler.dagId}\u0000${handler.taskId}`;
- if (
- entry?.kind === "handlers"
- ? entry.handlers.has(handler.taskId)
- : incomingTaskHandlers.has(pair)
- ) {
- throw new Error(
- `A handler for Dag "${handler.dagId}" task "${handler.taskId}" is
already registered`,
- );
- }
- incomingTaskHandlers.add(pair);
+ this.#dags = dags;
+ this.#taskHandlers = taskHandlers;
}
/**
@@ -218,12 +150,54 @@ export class Bundle {
/** Look up a registered handler, the way the runtime dispatches a task.
* Returns `undefined` when no handler exists. */
getTaskHandler(dagId: string, taskId: string): TaskFunction | undefined {
- const entry = this.#entries.get(dagId);
- if (entry === undefined) return undefined;
- return entry.kind === "dag"
- ? getDagTaskRecords(entry.dag).get(taskId)?.fn
- : entry.handlers.get(taskId);
+ const dag = this.#dags.get(dagId);
+ return dag === undefined
+ ? this.#taskHandlers.get(dagId)?.get(taskId)
+ : getDagTaskRecords(dag).get(taskId)?.fn;
+ }
+}
+
+// A Dag declared in TypeScript owns its own tasks, so a dag_id it holds and a
+// dag_id task handlers hold are two disagreeing sources for one task list.
+// Each kind is therefore rejected for a dag_id the other one already holds,
+// whichever order they were registered in.
+
+function stageDag(
+ dags: Map<string, Dag>,
+ taskHandlers: ReadonlyMap<string, ReadonlyMap<string, TaskFunction>>,
+ dag: Dag,
+): void {
+ if (taskHandlers.has(dag.dagId)) {
+ throw new Error(
+ `Dag "${dag.dagId}" already has registered task handlers; a Dag declared
in ` +
+ "TypeScript owns its own tasks, so one Dag ID cannot have both",
+ );
+ }
+ if (dags.has(dag.dagId)) {
+ throw new Error(`Dag "${dag.dagId}" is already registered`);
+ }
+ dags.set(dag.dagId, dag);
+}
+
+function stageTaskHandler(
+ dags: ReadonlyMap<string, Dag>,
+ taskHandlers: Map<string, Map<string, TaskFunction>>,
+ handler: TaskHandler,
+): void {
+ if (dags.has(handler.dagId)) {
+ throw new Error(
+ `Dag "${handler.dagId}" is declared in TypeScript; attach its tasks with
` +
+ "dag.task(...) rather than registering task handlers for them",
+ );
}
+ const forDag = taskHandlers.get(handler.dagId) ?? new Map<string,
TaskFunction>();
+ if (forDag.has(handler.taskId)) {
+ throw new Error(
+ `A handler for Dag "${handler.dagId}" task "${handler.taskId}" is
already registered`,
+ );
+ }
+ forDag.set(handler.taskId, getTaskHandlerFunction(handler));
+ taskHandlers.set(handler.dagId, forDag);
}
/** Internal: reject a `this` that is not a Bundle built by this copy, naming
@@ -237,21 +211,24 @@ export function validateOwnBundle(value: unknown,
accessor: string): asserts val
);
}
-/** Internal: every task handle this bundle can dispatch, across both kinds.
Not
- * re-exported from the package root: enumerating what the runtime dispatches
- * is the runtime's job. */
-export function listBundleTasks(bundle: Bundle): TaskRef[] {
- return [...entriesOf(bundle)].flatMap(([dagId, entry]) =>
- entryTaskIds(entry).map((taskId) => ({ dagId, taskId })),
- );
+/** Internal: finalize every Dag this bundle declared in TypeScript, so no task
+ * can be added or wired afterwards. */
+export function finalizeBundleDags(bundle: Bundle): void {
+ for (const dag of dagsOf(bundle).values()) {
+ finalizeDag(dag);
+ }
}
-/** Internal: every Dag this bundle provides for, with its task IDs. A native
- * Dag with no tasks is included; a Dag known only through task handlers
always
- * has at least one, since a handler is what put it here. */
-export function listBundleDags(bundle: Bundle): RegisteredDag[] {
- return [...entriesOf(bundle)].map(([dagId, entry]) => ({
- dagId,
- tasks: entryTaskIds(entry),
- }));
+/** Internal: the task IDs this bundle can dispatch, per Dag: the Dags declared
+ * in TypeScript first, then the Dags its task handlers name, each in
+ * registration order. A Dag declared in TypeScript with no tasks is
included. */
+export function bundleDagTaskIds(bundle: Bundle): Map<string, string[]> {
+ const byDag = new Map<string, string[]>();
+ for (const [dagId, dag] of dagsOf(bundle)) {
+ byDag.set(dagId, [...dag.taskIds]);
+ }
+ for (const [dagId, handlers] of taskHandlersOf(bundle)) {
+ byDag.set(dagId, [...handlers.keys()]);
+ }
+ return byDag;
}
diff --git a/ts-sdk/src/sdk/dag.ts b/ts-sdk/src/sdk/dag.ts
index 1da149473bb..94e2a45ec63 100644
--- a/ts-sdk/src/sdk/dag.ts
+++ b/ts-sdk/src/sdk/dag.ts
@@ -17,9 +17,13 @@
* under the License.
*/
-// The Dag authoring surface: `new Dag(dagId)` plus `dag.task(taskId,
handler)`.
+// The Dag authoring surface: `new Dag(dagId)`, `dag.task(taskId, handler)`,
and
+// the factory it returns. Calling a factory declares the task's place in the
+// Dag and supplies its arguments, the way calling a TaskFlow function does in
+// Python.
-import { brand, hasBrand } from "./brand.js";
+import { brand, DUPLICATE_COPY_HINT, hasBrand } from "./brand.js";
+import type { JsonValue } from "./client-types.js";
import type { TaskFunction } from "./task.js";
function isPlainRecord(value: unknown): value is Record<string, unknown> {
@@ -54,66 +58,195 @@ export type DagSpec = Record<string, never>;
*/
export type TaskSpec = Record<string, never>;
+// Carries a reference's return type without carrying a value. Not exported, so
+// the property cannot be read or written from outside; it exists only so
+// `TaskRef<boolean>` and `TaskRef<number>` are different types to the
compiler.
+declare const RETURN_TYPE: unique symbol;
+
/**
- * A reference to a task registered on a {@link Dag}, returned by
`dag.task(...)`.
+ * A reference to the result of one task, returned by calling that task.
+ *
+ * Identity only: the handler and the value are deliberately not exposed. Pass
a
+ * reference as an input of a downstream task to make that task depend on it.
*
- * Identity only: the handler is deliberately not exposed.
+ * `TReturn` is the handler's return type, so a construct that needs a
+ * particular one can ask for it. A reference of a narrower type is usable
+ * wherever a wider one is: a `TaskRef<number>` is a `TaskRef<unknown>`.
*
- * References are what `inputs` accepts, and what native Dag declaration will
- * use to wire dependencies.
+ * A reference carries a hidden brand, so a hand-written `{dagId, taskId}`
+ * object is not one: an input also takes a plain JSON value, and without the
+ * brand such an object could not be told apart from an upstream reference.
*/
-export interface TaskRef {
+export interface TaskRef<TReturn = unknown> {
/** Identifier of the Dag this task belongs to. */
readonly dagId: string;
/** Airflow task ID, including any TaskGroup prefix. */
readonly taskId: string;
+ /** @internal Never set; see {@link RETURN_TYPE}. */
+ readonly [RETURN_TYPE]?: TReturn;
+}
+
+/** Whether `value` is a TaskRef returned by any copy of this package. */
+function isTaskRef(value: unknown): value is TaskRef {
+ return hasBrand(value, "TaskRef");
+}
+
+/**
+ * The first reference reachable inside `value`, or `undefined` when there is
+ * none.
+ *
+ * A reference is an edge, and an edge has to be an input in its own right: one
+ * buried in an array or an object is data the serializer would write into the
+ * Dag verbatim, leaving the task running without the upstream it was given.
+ * TypeScript rejects that for a well-typed argument, so this catches the
`any`,
+ * the cast and the plain-JavaScript caller.
+ */
+function findNestedTaskRef(value: unknown, seen: WeakSet<object>): TaskRef |
undefined {
+ if (typeof value !== "object" || value === null) return undefined;
+ if (isTaskRef(value)) return value;
+ // A literal that refers back to itself is not JSON either, but it reaches
+ // here before anything else has rejected it.
+ if (seen.has(value)) return undefined;
+ seen.add(value);
+ for (const nested of Array.isArray(value) ? value : Object.values(value)) {
+ const found = findNestedTaskRef(nested, seen);
+ if (found) return found;
+ }
+ return undefined;
}
/**
- * Task references keyed by input name.
+ * The part of `T` a literal can express, matched structurally.
+ *
+ * `Extract<T, JsonValue>` would do for a type literal but collapses an
+ * `interface` to `never`, since an interface gets no implicit index signature
+ * and so never matches `JsonValue`'s object arm. An object survives only if
+ * every property does, which leaves types JSON cannot carry — a `Date`, a
+ * method — as `never`, so such an argument can only be given a reference.
+ */
+type JsonCompatible<T> = T extends JsonValue
+ ? T
+ : // A function is an object with no keys, so it would otherwise map to `{}`
+ // and take every method of a class along with it.
+ T extends (...args: never[]) => unknown
+ ? never
+ : T extends readonly (infer TElement)[]
+ ? readonly JsonCompatible<TElement>[]
+ : T extends object
+ ? T extends { [K in keyof T]: JsonCompatible<T[K]> }
+ ? { [K in keyof T]: JsonCompatible<T[K]> }
+ : never
+ : never;
+
+/**
+ * One input of a task: the upstream task that produces the value, or the
value.
+ *
+ * A literal is restricted to the JSON-compatible part of the argument's type,
+ * because it has to survive the trip through the serialized Dag. An argument
+ * that cannot be expressed as JSON at all, such as a `Date`, can only be given
+ * a reference.
+ */
+export type TaskInput<TValue> = TaskRef<TValue> | JsonCompatible<TValue>;
+
+/** The inputs of a task that declares several arguments, in declaration
order. */
+export type PositionalInputs<TParams extends readonly unknown[]> = {
+ [K in keyof TParams]: TaskInput<TParams[K]>;
+};
+
+/** The inputs of a task that declares one object of named arguments, by name.
*/
+export type TaskInputs<TArgs> = {
+ [K in keyof TArgs]: TaskRef | JsonCompatible<TArgs[K]>;
+};
+
+// Offered only where it means something: `TaskInputs<number>` would map over
+// `number`'s own methods and accept `{ toFixed: ... }`.
+type NamedInputs<TOnly> = [TOnly] extends [object] ? TaskInputs<TOnly> : never;
+
+/**
+ * What `dag.task(...)` returns: call it to declare where the task sits in the
Dag.
+ *
+ * Pass one input per argument the handler declares, in order. An input is
+ * either another task's reference, which makes this task wait for that task
and
+ * receive its result, or a literal value:
+ *
+ * ```ts
+ * const extract = dag.task("extract", async (): Promise<number> => 42);
+ * const transform = dag.task("transform", async (rows: number, region:
string) => rows);
+ * const load = dag.task("load", async (total: number) => {});
+ *
+ * load(transform(extract(), "us"));
+ * ```
+ *
+ * A handler that declares a single object of named arguments can also be
called
+ * with that object, which names each input instead of ordering it:
*
- * Stored and validated, but they do not create dependencies or pass values yet
- * (see {@link TaskOptions.inputs}). Each must identify an earlier task in the
- * same Dag. Literal values are not supported.
+ * ```ts
+ * const store = dag.task("store", async ({ total }: { total: number }) => {});
+ *
+ * store({ total: extract() });
+ * ```
+ *
+ * The compiler checks that every argument is supplied and that each literal
+ * matches its argument's type. A reference passed by position is checked
against
+ * the argument's type as well, which is what tells the two call shapes apart
+ * when a handler declares a single argument.
*/
-export type TaskInputs = Readonly<Record<string, TaskRef>>;
+export type TaskFactory<TParams extends readonly unknown[], TReturn = unknown>
= [TParams] extends [
+ readonly [],
+]
+ ? () => TaskRef<TReturn>
+ : TParams extends readonly [infer TOnly]
+ ? (input: TaskInput<TOnly> | NamedInputs<TOnly>) => TaskRef<TReturn>
+ : (...inputs: PositionalInputs<TParams>) => TaskRef<TReturn>;
/**
* Named options for `dag.task()`.
*
- * Keyword-only so neither field has to be positioned around the other, and so
- * future fields can be added without a new parameter. Unknown keys are
- * rejected, so a typo fails at import time rather than being ignored.
+ * Keyword-only so future fields can be added without a new parameter. Unknown
+ * keys are rejected, so a typo fails at import time rather than being ignored.
*/
export interface TaskOptions {
+ /** Task-level options. Stored, but not used yet — see {@link TaskSpec}. */
+ readonly spec?: TaskSpec;
/**
- * References to the upstream tasks this task consumes.
+ * Names for the handler's positional arguments, in declaration order.
*
- * Not used yet: a handler takes no arguments, and the Python stub Dag
- * defines task order. Read an upstream return value explicitly instead, with
- * `getClient().getXCom({ key: "return_value", taskId: "extract" })`, where
- * omitting `taskId` reads the *running* task's own XCom, not the upstream.
- *
- * In the future these will declare dependencies in native TypeScript Dags.
+ * `airflow-ts-pack` fills this in from the handler's parameter list, so the
+ * Dag names each argument as its handler does. Positional inputs bind by
+ * order, so a name left out only costs the label: `arg0`, `arg1` and so on
+ * stand in for it.
*/
- readonly inputs?: TaskInputs;
- /** Task-level options. Stored, but not used yet (see {@link TaskSpec}). */
- readonly spec?: TaskSpec;
+ readonly argNames?: readonly string[];
}
-/** Per-task record a Dag retains: the handle, the function, its spec, and the
- * upstream handles feeding it. */
+/** Per-task record a Dag retains: the reference, the handler, and its spec. */
export interface TaskRecord {
readonly task: TaskRef;
readonly fn: TaskFunction;
readonly spec: TaskSpec;
- /** Upstream handles keyed by input name; empty when the task has no inputs.
*/
- readonly inputs: TaskInputs;
}
-// Assigned inside Dag's static block: gives package-internal code read access
-// to the #tasks private field without a public accessor on the Dag class.
+/**
+ * Internal: what one call to a task factory recorded, by argument name.
+ *
+ * A {@link TaskRef} is an edge from the upstream task; anything else is a
+ * constant argument.
+ *
+ * Recorded for the serializer that will turn a native Dag into serialized Dag
+ * JSON; nothing reads them at execution time. What a running task is called
+ * with comes from the supervisor's `arg_bindings`, which are derived from the
+ * serialized Dag and resolved per task instance — see decision G of
+ * `airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`. Those
+ * bindings therefore win over anything recorded here, and the two never have
to
+ * be reconciled: a native Dag's wiring is what produced its bindings.
+ */
+export type RecordedInputs = Readonly<Record<string, TaskRef | JsonValue>>;
+
+// Assigned inside Dag's static block: gives package-internal code access to
+// Dag's private state without public accessors on the class.
let taskRecordsOf: (dag: Dag) => ReadonlyMap<string, TaskRecord>;
+let inputsOf: (dag: Dag) => ReadonlyMap<string, RecordedInputs>;
+let finalizeOf: (dag: Dag) => void;
/** Internal: whether `value` is a Dag built by any copy of this package. */
export function isDag(value: unknown): value is Dag {
@@ -123,33 +256,44 @@ export function isDag(value: unknown): value is Dag {
/**
* A Dag declared in TypeScript.
*
- * Today the Dag structure itself is still declared by a Python stub file; a
- * `Dag` instance binds TypeScript handlers to that stub's Dag/task IDs. The
- * instance retains its `spec` and every task's `(taskId, handler, spec)` so a
- * future `serialize()` can produce the serialized Dag JSON for native
- * TypeScript Dag declaration.
+ * Declare the tasks with `dag.task(taskId, handler)`, then call what each
+ * returns to lay the Dag out:
+ *
+ * ```ts
+ * const extract = dag.task("extract", async (): Promise<number> => 42);
+ * const load = dag.task("load", async ({ rows }: { rows: number }) => {});
+ * load({ rows: extract() });
+ * ```
+ *
+ * Every task has to be called exactly once, so none can be left out of the Dag
+ * by accident. A TypeScript handler for a task that a *Python* Dag declares is
+ * a `TaskHandler` instead, not a task on a `Dag`.
*
* Constructing a Dag has no effect beyond the instance itself. Collect the
ones
* a bundle should serve on a `Bundle` and await `bundle.serve()`.
*/
export class Dag {
- /** Identifier of this Dag. Must match the Python Dag's `dag_id`. */
+ /** Identifier of this Dag. */
readonly dagId: string;
/** Dag-level options this instance was constructed with, copied and frozen.
*/
readonly spec: DagSpec;
readonly #tasks = new Map<string, TaskRecord>();
+ readonly #inputs = new Map<string, RecordedInputs>();
+ #finalized = false;
static {
taskRecordsOf = (dag) => dag.#tasks;
+ inputsOf = (dag) => dag.#inputs;
+ finalizeOf = (dag) => dag.#finalize();
}
constructor(dagId: string, spec: DagSpec = {}) {
validateEmptySpec(`spec for Dag "${dagId}"`, spec);
brand(this, "Dag");
this.dagId = dagId;
- // Copied and frozen, as task specs and inputs are: nothing reads a spec
- // until the bundle manifest is built, long after the user's module has
run,
- // so a later mutation of their object would silently change what is
packed.
+ // Copied and frozen, as task specs are: nothing reads a spec until the
+ // bundle manifest is built, long after the user's module has run, so a
+ // later mutation of their object would silently change what is packed.
// Shallow, so a nested value in a future generated spec stays mutable.
this.spec = Object.freeze({ ...spec });
}
@@ -160,82 +304,215 @@ export class Dag {
}
/**
- * Register a TypeScript handler for a task of this Dag.
+ * Declare a task of this Dag, and return the factory that places it.
*
- * `taskId` must match the Dag-side operator's `task_id` exactly, including
- * any TaskGroup prefix. Returns this task's handle.
+ * Every argument the handler declares becomes an input of the returned
+ * {@link TaskFactory}; `getContext()` and `getClient()` reach the runtime
+ * from inside the call, so neither is an argument.
*/
- task<TArgs = void, TReturn = unknown>(
+ task<TParams extends readonly unknown[] = [], TReturn = unknown>(
taskId: string,
- handler: TaskFunction<TArgs, TReturn>,
+ handler: (...args: TParams) => TReturn | Promise<TReturn>,
options: TaskOptions = {},
- ): TaskRef {
+ ): TaskFactory<TParams, TReturn> {
if (typeof handler !== "function") {
throw new Error(`handler for Dag "${this.dagId}" task "${taskId}" must
be a function`);
}
if (this.#tasks.has(taskId)) {
throw new Error(`Task "${taskId}" is already registered for Dag
"${this.dagId}"`);
}
+ // A task added after the Dag was read could no longer be wired into it,
and
+ // would sit in the Dag unplaced and unreported.
+ if (this.#finalized) {
+ throw new Error(
+ `Task "${taskId}" cannot be added to Dag "${this.dagId}" after the Dag
was read; ` +
+ "declare every task while the module is loading",
+ );
+ }
this.#validateOptions(taskId, options);
- const { inputs = {}, spec = {} } = options;
+ const { spec = {} } = options;
validateEmptySpec(`spec for Dag "${this.dagId}" task "${taskId}"`, spec);
- this.#validateInputs(taskId, inputs);
- const task: TaskRef = Object.freeze({ dagId: this.dagId, taskId });
+ const argNames = this.#validateArgNames(taskId, options.argNames);
+ const task = createTaskRef(this.dagId, taskId);
this.#tasks.set(taskId, {
task,
- fn: handler as TaskFunction,
+ // The runtime dispatches every handler through one instantiation, as it
+ // does a registered TaskHandler; a positional one is wrapped at wiring.
+ fn: handler as unknown as TaskFunction,
spec: Object.freeze({ ...spec }),
- inputs: Object.freeze({ ...inputs }),
});
- return task;
+ return ((...inputs: unknown[]) => {
+ this.#wire(taskId, inputs, argNames);
+ return task;
+ }) as TaskFactory<TParams, TReturn>;
}
- // TypeScript is bypassable from plain JavaScript or an `as TaskOptions`
- // cast, so an unknown key is rejected rather than silently ignored.
+ // TypeScript is bypassable — from plain JavaScript, or an `as TaskOptions`
+ // cast — so an unknown key is rejected rather than silently ignored.
#validateOptions(taskId: string, options: TaskOptions): void {
const value: unknown = options;
if (!isPlainRecord(value)) {
throw new Error(`options for Dag "${this.dagId}" task "${taskId}" must
be an object`);
}
for (const key of Object.keys(value)) {
- if (key !== "inputs" && key !== "spec") {
+ if (key !== "spec" && key !== "argNames") {
throw new Error(`Unknown option "${key}" for Dag "${this.dagId}" task
"${taskId}"`);
}
}
}
- #validateInputs(taskId: string, inputs: TaskInputs): void {
- if (!isPlainRecord(inputs)) {
- throw new Error(`inputs for Dag "${this.dagId}" task "${taskId}" must be
an object`);
+ // TypeScript is bypassable, and these names become the keys the arguments
are
+ // recorded under, so an integer-like one would reorder what it labels.
+ #validateArgNames(taskId: string, names: unknown): readonly string[] |
undefined {
+ if (names === undefined) return undefined;
+ const describe = `argNames for Dag "${this.dagId}" task "${taskId}"`;
+ if (!Array.isArray(names)) throw new Error(`${describe} must be an array
of names`);
+ const seen = new Set<string>();
+ for (const name of names as unknown[]) {
+ if (typeof name !== "string" || name.length === 0 || /^\d+$/.test(name))
{
+ throw new Error(
+ `${describe} holds ${JSON.stringify(name)}; each name must be a
non-empty ` +
+ "string that is not a number",
+ );
+ }
+ if (seen.has(name)) throw new Error(`${describe} names "${name}" twice`);
+ seen.add(name);
+ }
+ return Object.freeze([...(names as string[])]);
+ }
+
+ #wire(taskId: string, inputs: readonly unknown[], argNames: readonly
string[] | undefined): void {
+ if (this.#finalized) {
+ throw new Error(
+ `Task "${taskId}" of Dag "${this.dagId}" was called after the Dag was
read; ` +
+ "call every task while the module is loading",
+ );
+ }
+ if (this.#inputs.has(taskId)) {
+ throw new Error(
+ `Task "${taskId}" of Dag "${this.dagId}" was already called; a task
holds one place ` +
+ "in a Dag, so call it once and reuse the reference",
+ );
}
- for (const [name, upstream] of Object.entries(inputs)) {
- if (
- upstream == null ||
- typeof upstream.dagId !== "string" ||
- typeof upstream.taskId !== "string"
- ) {
+ const positional = !isNamedCall(inputs);
+ const recorded = this.#checkInputs(
+ taskId,
+ positional ? positionalInputs(inputs, argNames) : (inputs[0] as
Record<string, unknown>),
+ );
+ this.#inputs.set(taskId, recorded);
+ if (positional && inputs.length > 0) {
+ const record = this.#tasks.get(taskId)!;
+ this.#tasks.set(taskId, { ...record, fn: spreadArgs(record.fn) });
+ }
+ }
+
+ #checkInputs(taskId: string, inputs: Record<string, unknown>):
RecordedInputs {
+ // Every own key, not just the enumerable string ones: a symbol key would
be
+ // copied by the spread below and so has to be checked, not stepped over.
+ for (const key of Reflect.ownKeys(inputs)) {
+ if (typeof key === "symbol") {
+ throw new Error(
+ `Input "${String(key)}" of task "${taskId}" is keyed by a symbol; ` +
+ "an argument name is a string",
+ );
+ }
+ const value = inputs[key];
+ // Anything unbranded is a literal argument, including a look-alike
+ // `{dagId, taskId}` object: only a real reference makes an edge.
+ if (!isTaskRef(value)) {
+ const nested = findNestedTaskRef(value, new WeakSet());
+ if (nested) {
+ throw new Error(
+ `Input "${key}" of task "${taskId}" holds a reference to
"${nested.taskId}" inside a ` +
+ "literal value, which draws no edge; pass the reference as the
input itself, or " +
+ "give each upstream its own argument",
+ );
+ }
+ continue;
+ }
+ if (value.dagId !== this.dagId) {
+ throw new Error(
+ `Input "${key}" of task "${taskId}" comes from Dag "${value.dagId}",
not "${this.dagId}"`,
+ );
+ }
+ const upstream = this.#tasks.get(value.taskId);
+ if (!upstream) {
throw new Error(
- `Input "${name}" of task "${taskId}" must be a task handle returned
by dag.task(...)`,
+ `Input "${key}" of task "${taskId}" refers to unregistered task
"${value.taskId}"`,
);
}
- if (upstream.dagId !== this.dagId) {
+ // Identity, not just the ID pair: two Dag objects can carry the same
+ // dagId, and a second resolved copy of this package brands its own
+ // references, so matching IDs do not make a reference this Dag handed
out.
+ if (upstream.task !== value) {
throw new Error(
- `Input "${name}" of task "${taskId}" comes from Dag
"${upstream.dagId}", not "${this.dagId}"`,
+ `Input "${key}" of task "${taskId}" was not returned by this Dag's
"${value.taskId}"; ` +
+ `it comes from another Dag object with the same ID, or
${DUPLICATE_COPY_HINT}`,
);
}
- // An input can only name a task registered earlier on this Dag, which
- // makes self-references and cycles unrepresentable.
- if (!this.#tasks.has(upstream.taskId)) {
+ }
+ return Object.freeze({ ...inputs }) as RecordedInputs;
+ }
+
+ #finalize(): void {
+ if (this.#finalized) return;
+ for (const taskId of this.#tasks.keys()) {
+ if (!this.#inputs.has(taskId)) {
throw new Error(
- `Input "${name}" of task "${taskId}" refers to unregistered task
"${upstream.taskId}"`,
+ `Task "${taskId}" of Dag "${this.dagId}" is never called, so it has
no place in the ` +
+ "Dag; call the factory dag.task(...) returned",
);
}
}
+ // Last, so a Dag that failed the check reports that same failure again
+ // rather than reporting itself as already read.
+ this.#finalized = true;
}
}
/**
- * Internal: the task records of a Dag, for registry lookups.
+ * Whether a call named its inputs rather than ordering them.
+ *
+ * One plain object is the named form. A reference is a plain object too, so it
+ * is ruled out first: `load(extract())` is one positional input, not a map of
+ * argument names.
+ */
+function isNamedCall(inputs: readonly unknown[]): boolean {
+ return inputs.length === 1 && !isTaskRef(inputs[0]) &&
isPlainRecord(inputs[0]);
+}
+
+function positionalInputs(
+ inputs: readonly unknown[],
+ argNames: readonly string[] | undefined,
+): Record<string, unknown> {
+ const byName: Record<string, unknown> = {};
+ inputs.forEach((value, index) => {
+ byName[argNames?.[index] ?? `arg${index}`] = value;
+ });
+ return byName;
+}
+
+/**
+ * Dispatch a positional handler through the one call shape the runtime uses.
+ *
+ * A task is called with its bound arguments as a single object, in the order
+ * they were recorded, so spreading its values back restores the argument list
+ * the handler declared.
+ */
+function spreadArgs(fn: TaskFunction): TaskFunction {
+ const handler = fn as unknown as (...args: unknown[]) => unknown;
+ return ((args: Record<string, unknown>) =>
+ handler(...Object.values(args ?? {}))) as unknown as TaskFunction;
+}
+
+function createTaskRef(dagId: string, taskId: string): TaskRef {
+ const task: TaskRef = { dagId, taskId };
+ brand(task, "TaskRef");
+ return Object.freeze(task);
+}
+
+/**
+ * Internal: the task records of a Dag, for bundle lookups.
*
* Not re-exported from the package root, and the package `"exports"` map
* blocks deep imports, so this is unreachable from outside the SDK.
@@ -243,3 +520,19 @@ export class Dag {
export function getDagTaskRecords(dag: Dag): ReadonlyMap<string, TaskRecord> {
return taskRecordsOf(dag);
}
+
+/** Internal: what each task of a Dag was called with, keyed by task ID.
+ * A task that has not been called is absent. */
+export function getDagTaskInputs(dag: Dag): ReadonlyMap<string,
RecordedInputs> {
+ return inputsOf(dag);
+}
+
+/**
+ * Internal: check that `dag` is fully laid out, then finalize it against
further wiring.
+ *
+ * Idempotent, and never part of the public surface: a Dag is finished when its
+ * module has finished loading, so there is nothing for an author to call.
+ */
+export function finalizeDag(dag: Dag): void {
+ finalizeOf(dag);
+}
diff --git a/ts-sdk/tests/cli/fixtures/entry.ts
b/ts-sdk/tests/cli/fixtures/entry.ts
index 607c62cdf0b..77fe93fcd9a 100644
--- a/ts-sdk/tests/cli/fixtures/entry.ts
+++ b/ts-sdk/tests/cli/fixtures/entry.ts
@@ -22,7 +22,7 @@ import { Bundle, Dag, TaskHandler } from
"../../../src/index.js";
// A mixed bundle: task handlers for a Python-owned Dag, plus a natively
// declared one, so packing covers both paths into the manifest.
const otherDag = new Dag("other_dag");
-otherDag.task("solo", async () => undefined);
+otherDag.task("solo", async () => undefined)();
await new Bundle(
new TaskHandler("fixture_dag", "extract", async () => "extracted"),
diff --git a/ts-sdk/tests/cli/fixtures/noisy-entry.ts
b/ts-sdk/tests/cli/fixtures/noisy-entry.ts
index 370022287c7..ed7ab0683c3 100644
--- a/ts-sdk/tests/cli/fixtures/noisy-entry.ts
+++ b/ts-sdk/tests/cli/fixtures/noisy-entry.ts
@@ -23,6 +23,6 @@ import { Bundle, Dag } from "../../../src/index.js";
console.log("noise from an import-time dependency");
const noisyDag = new Dag("noisy_dag");
-noisyDag.task("only", async () => undefined);
+noisyDag.task("only", async () => undefined)();
await new Bundle(noisyDag).serve();
diff --git a/ts-sdk/tests/cli/pack.test.ts b/ts-sdk/tests/cli/pack.test.ts
index fad4d50a8a8..c9d8e6a9ea6 100644
--- a/ts-sdk/tests/cli/pack.test.ts
+++ b/ts-sdk/tests/cli/pack.test.ts
@@ -427,7 +427,7 @@ describe("runPack", () => {
[
`import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
'const bigDag = new Dag("big_dag");',
- 'for (let i = 0; i < 5000; i += 1) bigDag.task(String(i).padStart(240,
"t"), async () => undefined);',
+ 'for (let i = 0; i < 5000; i += 1) bigDag.task(String(i).padStart(240,
"t"), async () => undefined)();',
"await new Bundle(bigDag).serve();",
].join("\n"),
);
@@ -472,7 +472,7 @@ describe("runPack", () => {
[
`import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
`const suspiciousDag = new Dag(${JSON.stringify(dagId)});`,
- `suspiciousDag.task(${JSON.stringify(taskId)}, async () =>
undefined);`,
+ `suspiciousDag.task(${JSON.stringify(taskId)}, async () =>
undefined)();`,
"await new Bundle(suspiciousDag).serve();",
].join("\n"),
);
@@ -542,7 +542,7 @@ describe("runPack", () => {
[
`import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
'const salesDag = new Dag("sales_dag");',
- 'salesDag.task("extract", async () => undefined);',
+ 'salesDag.task("extract", async () => undefined)();',
'await new Bundle(salesDag, new Dag("empty_dag")).serve();',
].join("\n"),
);
@@ -565,9 +565,9 @@ describe("runPack", () => {
[
`import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
'const salesDag = new Dag("sales_dag");',
- 'salesDag.task("extract", async () => undefined);',
+ 'salesDag.task("extract", async () => undefined)();',
'const billingDag = new Dag("billing_dag");',
- 'billingDag.task("charge", async () => undefined);',
+ 'billingDag.task("charge", async () => undefined)();',
"await new Bundle(salesDag).serve();",
].join("\n"),
);
diff --git a/ts-sdk/tests/coordinator/integration.test.ts
b/ts-sdk/tests/coordinator/integration.test.ts
index c5a6ffc1eaf..d0b6d53f83f 100644
--- a/ts-sdk/tests/coordinator/integration.test.ts
+++ b/ts-sdk/tests/coordinator/integration.test.ts
@@ -28,7 +28,7 @@
// No Python, no Airflow install, but exercises the same wire format
// the real coordinator speaks.
-import { afterEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { createServer, Socket as NetSocket, type Server, type Socket } from
"node:net";
import { encode, decode } from "@msgpack/msgpack";
import {
@@ -49,13 +49,22 @@ interface BoundTransformArgs {
dryRun: boolean;
}
-const testDag = new Dag("test_dag");
-const otherDag = new Dag("other_dag");
// The bundle the runtime dispatches through. startCoordinator() is driven
// directly rather than through bundle.serve(), so these tests can supply mock
// socket addresses. Both authoring kinds are registered in one call, since the
// runtime dispatches through one lookup regardless of which put a task there.
-const bundle = new Bundle(testDag, otherDag);
+//
+// Rebuilt per test: dispatching reads the bundle, which finalizes every native
+// Dag in it against further tasks, so one Dag cannot collect a task per test.
+let testDag: Dag;
+let otherDag: Dag;
+let bundle: Bundle;
+
+beforeEach(() => {
+ testDag = new Dag("test_dag");
+ otherDag = new Dag("other_dag");
+ bundle = new Bundle(testDag, otherDag);
+});
interface MockResult {
firstResponse: { id: number; body: unknown; isResponse: boolean } | null;
@@ -254,7 +263,7 @@ describe("coordinator runtime integration", () => {
testDag.task("say_hello", async () => {
observedCtx = getContext();
return "ok";
- });
+ })();
const result = await driveSupervisor(makeStartupDetails("say_hello"));
@@ -307,7 +316,7 @@ describe("coordinator runtime integration", () => {
const commAccept = acceptOne(comm.server);
const logsAccept = acceptOne(logs.server);
- testDag.task("terminal_timeout", async () => undefined);
+ testDag.task("terminal_timeout", async () => undefined)();
const runtimeDone = startCoordinator(bundle, {
commAddr: `127.0.0.1:${comm.port}`,
logsAddr: `127.0.0.1:${logs.port}`,
@@ -339,7 +348,7 @@ describe("coordinator runtime integration", () => {
it("returns TaskState=failed when the handler throws", async () => {
testDag.task("boom", async () => {
throw new Error("boom");
- });
+ })();
const result = await driveSupervisor(makeStartupDetails("boom"));
@@ -352,7 +361,7 @@ describe("coordinator runtime integration", () => {
it("returns RetryTask when the handler throws and Airflow says the failure
is retryable", async () => {
testDag.task("boom_retry", async () => {
throw new Error("boom");
- });
+ })();
const result = await driveSupervisor(
makeStartupDetails("boom_retry", "test_dag", "r1", {
@@ -608,7 +617,7 @@ describe("coordinator runtime integration", () => {
process.emit("SIGTERM");
sawAbort = getContext().signal.aborted;
throw new Error("interrupted");
- });
+ })();
const result = await
driveSupervisor(makeStartupDetails("aborted_then_failed"));
@@ -626,7 +635,7 @@ describe("coordinator runtime integration", () => {
process.emit("SIGTERM");
sawAbort = getContext().signal.aborted;
throw new Error("interrupted");
- });
+ })();
const result = await driveSupervisor(
makeStartupDetails("aborted_then_failed_retry", "test_dag", "r1", {
@@ -649,7 +658,7 @@ describe("coordinator runtime integration", () => {
process.emit("SIGTERM");
sawAbort = getContext().signal.aborted;
return "completed";
- });
+ })();
const result = await
driveSupervisor(makeStartupDetails("completed_after_sigterm"));
@@ -703,7 +712,7 @@ describe("coordinator runtime integration", () => {
if (back !== `node says: ${observedGreeting}`) {
throw new Error(`xcom round-trip mismatch: ${back}`);
}
- });
+ })();
const responder: Responder = (msgType, body) => {
if (msgType === "GetVariable") {
@@ -753,7 +762,7 @@ describe("coordinator runtime integration", () => {
let observed: string | null = "<unset>";
testDag.task("missing_variable", async () => {
observed = await getClient().getVariable("missing_key");
- });
+ })();
const responder: Responder = (msgType) => {
if (msgType === "GetVariable") {
@@ -778,10 +787,10 @@ describe("coordinator runtime integration", () => {
let calledSecondDag = false;
testDag.task("shared_task", async () => {
calledFirstDag = true;
- });
+ })();
otherDag.task("shared_task", async () => {
calledSecondDag = true;
- });
+ })();
await driveSupervisor(makeStartupDetails("shared_task"));
@@ -804,7 +813,7 @@ describe("coordinator runtime integration", () => {
});
it("auto-pushes return_value XCom when handler returns a value", async () =>
{
- testDag.task("pusher", async () => "my-result");
+ testDag.task("pusher", async () => "my-result")();
const responder: Responder = (msgType, _body) => {
if (msgType === "SetXCom") return { body: null };
@@ -826,7 +835,7 @@ describe("coordinator runtime integration", () => {
it("does NOT push return_value XCom when handler returns undefined", async
() => {
testDag.task("void_task", async () => {
// no return value
- });
+ })();
const result = await driveSupervisor(makeStartupDetails("void_task"));
diff --git a/ts-sdk/tests/coordinator/runtime-manifest.test.ts
b/ts-sdk/tests/coordinator/runtime-manifest.test.ts
index b7e030f43e7..afca723af21 100644
--- a/ts-sdk/tests/coordinator/runtime-manifest.test.ts
+++ b/ts-sdk/tests/coordinator/runtime-manifest.test.ts
@@ -28,7 +28,7 @@ import { Bundle } from "../../src/sdk/bundle.js";
function buildDag(dagId: string, ...taskIds: string[]): Dag {
const dag = new Dag(dagId);
for (const taskId of taskIds) {
- dag.task(taskId, async () => undefined);
+ dag.task(taskId, async () => undefined)();
}
return dag;
}
@@ -87,7 +87,7 @@ describe("buildBundleManifest", () => {
it("rejects a non-string dagId before object-key coercion hides it", () => {
const dag = new Dag(123 as unknown as string);
- dag.task("t1", async () => undefined);
+ dag.task("t1", async () => undefined)();
expect(() => buildBundleManifest(new Bundle(dag))).toThrowError(/Dag ID
must be a string/);
});
});
diff --git a/ts-sdk/tests/public-api.test.ts b/ts-sdk/tests/public-api.test.ts
index 2a86f09d669..b62075579f1 100644
--- a/ts-sdk/tests/public-api.test.ts
+++ b/ts-sdk/tests/public-api.test.ts
@@ -27,8 +27,11 @@ import type {
SetXComOpts,
TaskClient,
Registerable,
+ PositionalInputs,
TaskContext,
+ TaskFactory,
TaskFunction,
+ TaskInput,
TaskInputs,
TaskOptions,
TaskRef,
@@ -50,10 +53,13 @@ import {
describe("public API", () => {
it("exports the Dag authoring surface", async () => {
const dag = new Dag("public_api_dag");
- const upstream = dag.task("public_api_task", async () => undefined);
- const downstream = dag.task("public_api_downstream", async () =>
undefined, {
- inputs: { upstream },
- });
+ const upstreamTask = dag.task("public_api_task", async () => undefined);
+ const downstreamTask = dag.task(
+ "public_api_downstream",
+ async (_: { upstream: undefined }) => undefined,
+ );
+ const upstream = upstreamTask();
+ const downstream = downstreamTask({ upstream });
expect(upstream).toEqual({ dagId: "public_api_dag", taskId:
"public_api_task" });
expect(downstream).toEqual({ dagId: "public_api_dag", taskId:
"public_api_downstream" });
expect(dag.taskIds).toEqual(["public_api_task", "public_api_downstream"]);
@@ -278,22 +284,41 @@ describe("public API", () => {
});
it("keeps the Dag authoring signatures extensible via trailing specs", () =>
{
- expectTypeOf<TaskRef>().toEqualTypeOf<{
- readonly dagId: string;
- readonly taskId: string;
- }>();
- expectTypeOf<TaskInputs>().toEqualTypeOf<Readonly<Record<string,
TaskRef>>>();
+ expectTypeOf<TaskRef["dagId"]>().toEqualTypeOf<string>();
+ expectTypeOf<TaskRef["taskId"]>().toEqualTypeOf<string>();
+ // A reference carries its handler's return type, so a construct that needs
+ // a particular one can ask for it: a narrower reference is usable wherever
+ // a wider one is, and not the other way round.
+ expectTypeOf<TaskRef<boolean>>().toMatchTypeOf<TaskRef>();
+ expectTypeOf<TaskRef>().not.toMatchTypeOf<TaskRef<boolean>>();
+ // Wiring moved to the factory call, so `inputs` is no longer an option and
+ // the only remaining one is the spec.
expectTypeOf<TaskOptions>().toEqualTypeOf<{
- readonly inputs?: TaskInputs;
readonly spec?: TaskSpec;
+ readonly argNames?: readonly string[];
}>();
+ // Each named argument takes any upstream reference or a literal of its
own type.
+ expectTypeOf<TaskInputs<{ rows: number }>>().toEqualTypeOf<{ rows: TaskRef
| number }>();
+ // A positional argument takes a literal or a reference of the argument's
own
+ // type, which is what tells a one-argument positional call from a named
one.
+ expectTypeOf<TaskInput<number>>().toEqualTypeOf<TaskRef<number> |
number>();
+ expectTypeOf<PositionalInputs<[number, string]>>().toEqualTypeOf<
+ [TaskRef<number> | number, TaskRef<string> | string]
+ >();
+ // A handler with no arguments is called with none; one with several is
+ // called with a value per argument, in order.
+ expectTypeOf<TaskFactory<[]>>().toEqualTypeOf<() => TaskRef>();
+ expectTypeOf<TaskFactory<[], boolean>>().toEqualTypeOf<() =>
TaskRef<boolean>>();
+ expectTypeOf<TaskFactory<[number, string]>>().toEqualTypeOf<
+ (...inputs: [TaskRef<number> | number, TaskRef<string> | string]) =>
TaskRef
+ >();
expectTypeOf<ConstructorParameters<typeof Dag>>().toEqualTypeOf<[string,
DagSpec?]>();
expectTypeOf<Dag["task"]>().toEqualTypeOf<
- <TArgs = void, TReturn = unknown>(
+ <TParams extends readonly unknown[] = [], TReturn = unknown>(
taskId: string,
- handler: TaskFunction<TArgs, TReturn>,
+ handler: (...args: TParams) => TReturn | Promise<TReturn>,
options?: TaskOptions,
- ) => TaskRef
+ ) => TaskFactory<TParams, TReturn>
>();
expectTypeOf<Dag["taskIds"]>().toEqualTypeOf<readonly string[]>();
// Reserved with no fields yet, so only `{}` is expressible. Generated
specs
@@ -391,21 +416,43 @@ describe("public API", () => {
// @ts-expect-error a task handler is required.
new Dag("example").task("extract");
const dag = new Dag("example");
- const upstream = dag.task("extract", async () => undefined);
- // @ts-expect-error inputs must be task handles, not arbitrary values.
+ const extract = dag.task("extract", async () => undefined);
+ // @ts-expect-error wiring belongs to the factory call, not the options.
dag.task("transform", async () => undefined, { inputs: { count: 1 } });
- // @ts-expect-error inputs and spec are keyword-only, not positional.
- dag.task("transform2", async () => undefined, { upstream });
+ // @ts-expect-error the spec is keyword-only, not positional.
+ dag.task("transform2", async () => undefined, { extract });
// @ts-expect-error a Dag spec is an options object, not a primitive.
new Dag("spec_dag", 42);
// @ts-expect-error DagSpec has no fields yet, so a schedule cannot be
declared here.
new Dag("spec_dag", { schedule: "@daily" });
// @ts-expect-error TaskSpec has no fields yet, so retries cannot be
declared here.
dag.task("transform3", async () => undefined, { spec: { retries: 2 } });
- // @ts-expect-error the TaskRef handle is data, not callable.
- upstream();
- // @ts-expect-error a bundle is built from Dags, not from task handles.
- new Bundle(upstream);
+ // @ts-expect-error a handler with no arguments is called with none.
+ extract({ rows: 1 });
+ const transform = dag.task("transform4", async (_: { rows: number }) =>
undefined);
+ // @ts-expect-error every argument the handler declares has to be
supplied.
+ transform({});
+ // @ts-expect-error a literal has to match its argument's type.
+ transform({ rows: "many" });
+ const totals = dag.task("totals", async (): Promise<{ rows: number }> =>
({ rows: 1 }));
+ // A single object of named arguments is given by name or by position.
+ transform({ rows: 1 });
+ transform(totals());
+ // @ts-expect-error a positional reference has to return the argument's
type.
+ transform(extract());
+ const pair = dag.task("pair", async (rows: number, region: string) =>
`${region}${rows}`);
+ pair(1, "us");
+ // @ts-expect-error a positional argument cannot be skipped.
+ pair(1);
+ // @ts-expect-error each positional literal has to match its own
argument.
+ pair("many", "us");
+ const stamp = dag.task("stamp", async (_: { at: Date }) => undefined);
+ // @ts-expect-error a Date cannot survive the serialized Dag, so only a
reference will do.
+ stamp({ at: new Date() });
+ // The reference an upstream call returns is always accepted.
+ stamp({ at: extract() });
+ // @ts-expect-error a bundle is built from Dags, not from task factories.
+ new Bundle(extract);
// @ts-expect-error serve() takes nothing; the bundle already holds it
all.
new Bundle(dag).serve(dag);
};
diff --git a/ts-sdk/tests/sdk/bundle.test.ts b/ts-sdk/tests/sdk/bundle.test.ts
index 37f6f68b663..24998153c27 100644
--- a/ts-sdk/tests/sdk/bundle.test.ts
+++ b/ts-sdk/tests/sdk/bundle.test.ts
@@ -19,13 +19,13 @@
import { describe, it, expect } from "vitest";
import { Dag } from "../../src/sdk/dag.js";
-import { Bundle, listBundleDags, listBundleTasks } from
"../../src/sdk/bundle.js";
+import { Bundle, bundleDagTaskIds, finalizeBundleDags } from
"../../src/sdk/bundle.js";
describe("Bundle", () => {
it("registers a Dag and retrieves its handlers", () => {
const handler = async () => "hello";
const dag = new Dag("example_dag");
- dag.task("my_task", handler);
+ dag.task("my_task", handler)();
const bundle = new Bundle();
bundle.register(dag);
expect(bundle.getTaskHandler("example_dag", "my_task")).toBe(handler);
@@ -34,13 +34,15 @@ describe("Bundle", () => {
it("registers the Dags passed to its constructor", () => {
const handler = async () => "hello";
const dagA = new Dag("dag_a");
- dagA.task("a", handler);
+ dagA.task("a", handler)();
const bundle = new Bundle(dagA, new Dag("dag_b"));
expect(bundle.getTaskHandler("dag_a", "a")).toBe(handler);
- expect(listBundleDags(bundle)).toEqual([
- { dagId: "dag_a", tasks: ["a"] },
- { dagId: "dag_b", tasks: [] },
- ]);
+ expect(bundleDagTaskIds(bundle)).toEqual(
+ new Map([
+ ["dag_a", ["a"]],
+ ["dag_b", []],
+ ]),
+ );
});
it("rejects duplicate dagIds passed to the constructor", () => {
@@ -58,28 +60,30 @@ describe("Bundle", () => {
it("returns undefined for unknown taskIds and dagIds", () => {
const bundle = new Bundle();
const dag = new Dag("example_dag");
- dag.task("my_task", async () => undefined);
+ dag.task("my_task", async () => undefined)();
bundle.register(dag);
expect(bundle.getTaskHandler("example_dag", "nope")).toBeUndefined();
expect(bundle.getTaskHandler("unknown_dag", "my_task")).toBeUndefined();
});
- it("returns an empty list when no Dags are registered", () => {
+ it("returns nothing when no Dags are registered", () => {
const bundle = new Bundle();
- expect(listBundleTasks(bundle)).toEqual([]);
+ expect(bundleDagTaskIds(bundle)).toEqual(new Map());
});
it("lists tasks across registered Dags", () => {
const dagA = new Dag("dag_a");
- dagA.task("a", async () => undefined);
+ dagA.task("a", async () => undefined)();
const dagB = new Dag("dag_b");
- dagB.task("b", async () => undefined);
+ dagB.task("b", async () => undefined)();
const bundle = new Bundle();
bundle.register(dagA, dagB);
- const registered = listBundleTasks(bundle);
- expect(registered).toHaveLength(2);
- expect(registered).toContainEqual({ dagId: "dag_a", taskId: "a" });
- expect(registered).toContainEqual({ dagId: "dag_b", taskId: "b" });
+ expect(bundleDagTaskIds(bundle)).toEqual(
+ new Map([
+ ["dag_a", ["a"]],
+ ["dag_b", ["b"]],
+ ]),
+ );
});
it("rejects registering the same dagId in separate calls", () => {
@@ -105,10 +109,10 @@ describe("Bundle", () => {
it("registers none of the Dags when a call throws", () => {
const bundle = new Bundle();
const dag = new Dag("dag_a");
- dag.task("a", async () => undefined);
+ dag.task("a", async () => undefined)();
expect(() => bundle.register(dag, new Dag("dag_a"))).toThrowError(/already
registered/);
expect(bundle.getTaskHandler("dag_a", "a")).toBeUndefined();
- expect(listBundleTasks(bundle)).toEqual([]);
+ expect(bundleDagTaskIds(bundle)).toEqual(new Map());
});
it("rejects values that are neither a Dag nor a task handler", () => {
@@ -129,14 +133,16 @@ describe("Bundle", () => {
it("lists every registered Dag with its tasks, empty Dags included", () => {
const dagA = new Dag("dag_a");
- dagA.task("a1", async () => undefined);
- dagA.task("a2", async () => undefined);
+ dagA.task("a1", async () => undefined)();
+ dagA.task("a2", async () => undefined)();
const bundle = new Bundle();
bundle.register(dagA, new Dag("empty_dag"));
- expect(listBundleDags(bundle)).toEqual([
- { dagId: "dag_a", tasks: ["a1", "a2"] },
- { dagId: "empty_dag", tasks: [] },
- ]);
+ expect(bundleDagTaskIds(bundle)).toEqual(
+ new Map([
+ ["dag_a", ["a1", "a2"]],
+ ["empty_dag", []],
+ ]),
+ );
});
it("accepts a call that registers nothing", () => {
@@ -144,7 +150,7 @@ describe("Bundle", () => {
// guard at the call site.
const bundle = new Bundle();
expect(() => bundle.register()).not.toThrow();
- expect(listBundleDags(bundle)).toEqual([]);
+ expect(bundleDagTaskIds(bundle)).toEqual(new Map());
});
it("carries the brand its own serve guard reads", () => {
@@ -153,18 +159,30 @@ describe("Bundle", () => {
expect(Symbol.for("airflow.ts-sdk.Bundle") in new Bundle()).toBe(true);
});
- it("sees tasks added to a Dag after registration", () => {
+ it("sees tasks added to a Dag between registration and the first read", ()
=> {
+ // Registration records Dag identity rather than a snapshot of its tasks,
so
+ // a Dag assembled across several modules is still complete when it is
read.
const bundle = new Bundle();
const dag = new Dag("example_dag");
bundle.register(dag);
- expect(listBundleTasks(bundle)).toEqual([]);
const handler = async () => "late";
- dag.task("late_task", handler);
+ dag.task("late_task", handler)();
expect(bundle.getTaskHandler("example_dag", "late_task")).toBe(handler);
- expect(listBundleTasks(bundle)).toContainEqual({
- dagId: "example_dag",
- taskId: "late_task",
- });
+ expect(bundleDagTaskIds(bundle).get("example_dag")).toContain("late_task");
+ });
+
+ it("rejects a task added to a Dag the bundle has already reported", () => {
+ const dag = new Dag("example_dag");
+ dag.task("extract", async () => undefined)();
+ const bundle = new Bundle(dag);
+ // Reporting what the bundle provides is what finalizes a native Dag;
+ // enumerating what it can dispatch does not.
+ expect(bundleDagTaskIds(bundle)).toEqual(new Map([["example_dag",
["extract"]]]));
+ finalizeBundleDags(bundle);
+
+ expect(() => dag.task("late_task", async () => "late")).toThrowError(
+ /Task "late_task" cannot be added to Dag "example_dag" after the Dag was
read/,
+ );
});
});
diff --git a/ts-sdk/tests/sdk/dag.test.ts b/ts-sdk/tests/sdk/dag.test.ts
index 4dfe1652652..5c772eb57c8 100644
--- a/ts-sdk/tests/sdk/dag.test.ts
+++ b/ts-sdk/tests/sdk/dag.test.ts
@@ -18,53 +18,104 @@
*/
import { describe, it, expect } from "vitest";
-import { Dag, getDagTaskRecords, type TaskRef } from "../../src/sdk/dag.js";
-import { Bundle } from "../../src/sdk/bundle.js";
+import {
+ Dag,
+ finalizeDag,
+ getDagTaskInputs,
+ getDagTaskRecords,
+ type TaskRef,
+} from "../../src/sdk/dag.js";
+import { Bundle, finalizeBundleDags } from "../../src/sdk/bundle.js";
describe("Dag", () => {
- it("returns a frozen TaskRef handle with the Dag and task identity", () => {
+ it("returns a factory whose call yields a frozen TaskRef with the Dag and
task identity", () => {
const dag = new Dag("example_dag");
- const task = dag.task("my_task", async () => "hello");
- expect(task).toEqual({ dagId: "example_dag", taskId: "my_task" });
- expect(Object.isFrozen(task)).toBe(true);
+ const myTask = dag.task("my_task", async () => "hello");
+ expect(typeof myTask).toBe("function");
+
+ const ref = myTask();
+ expect(ref).toEqual({ dagId: "example_dag", taskId: "my_task" });
+ expect(Object.isFrozen(ref)).toBe(true);
});
- it("chains upstream handles into downstream task inputs", () => {
+ it("chains upstream references into downstream task inputs", () => {
const dag = new Dag("chained_dag");
- const extracted = dag.task("extract", async () => ({ rows: 1 }));
- const transformed = dag.task("transform", async () => undefined, { inputs:
{ extracted } });
- const loaded = dag.task("load", async () => undefined, { inputs: {
transformed }, spec: {} });
+ const extract = dag.task("extract", async () => ({ rows: 1 }));
+ const transform = dag.task(
+ "transform",
+ async (_: { extracted: { rows: number } }) => undefined,
+ );
+ const load = dag.task("load", async (_: { transformed: undefined }) =>
undefined, { spec: {} });
+
+ const extracted = extract();
+ const transformed = transform({ extracted });
+ const loaded = load({ transformed });
expect(extracted).toEqual({ dagId: "chained_dag", taskId: "extract" });
expect(transformed).toEqual({ dagId: "chained_dag", taskId: "transform" });
expect(loaded).toEqual({ dagId: "chained_dag", taskId: "load" });
- const records = getDagTaskRecords(dag);
- expect(records.get("extract")?.inputs).toEqual({});
- expect(records.get("transform")?.inputs).toEqual({ extracted });
- expect(records.get("load")?.inputs).toEqual({ transformed });
+ const inputs = getDagTaskInputs(dag);
+ expect(inputs.get("extract")).toEqual({});
+ expect(inputs.get("transform")).toEqual({ extracted });
+ expect(inputs.get("load")).toEqual({ transformed });
});
- it("accepts several named inputs for one task", () => {
- const dag = new Dag("fan_in_dag");
- const extracted = dag.task("extract", async () => undefined);
- const otherTaskResult = dag.task("other_task", async () => undefined);
- dag.task("transform", async () => undefined, { inputs: { extracted,
otherTaskResult } });
+ it("records a literal argument as a value rather than an edge", () => {
+ const dag = new Dag("literal_dag");
+ const extract = dag.task("extract", async () => 1);
+ const transform = dag.task(
+ "transform",
+ async (_: { extracted: number; regionCode: string; limits: number[] })
=> undefined,
+ );
+
+ const extracted = extract();
+ transform({ extracted, regionCode: "us", limits: [1, 2] });
- expect(getDagTaskRecords(dag).get("transform")?.inputs).toEqual({
+ expect(getDagTaskInputs(dag).get("transform")).toEqual({
extracted,
- otherTaskResult,
+ regionCode: "us",
+ limits: [1, 2],
});
});
+ it("treats an unbranded look-alike reference as a literal, not an edge", ()
=> {
+ const dag = new Dag("lookalike_dag");
+ const transform = dag.task("transform", async (_: { upstream: unknown })
=> undefined);
+ const lookalike = { dagId: "lookalike_dag", taskId: "ghost" };
+
+ transform({ upstream: lookalike });
+
+ expect(getDagTaskInputs(dag).get("transform")).toEqual({ upstream:
lookalike });
+ });
+
+ it("accepts several named inputs for one task", () => {
+ const dag = new Dag("fan_in_dag");
+ const extract = dag.task("extract", async () => undefined);
+ const other = dag.task("other_task", async () => undefined);
+ const transform = dag.task(
+ "transform",
+ async (_: { extracted: undefined; otherTaskResult: undefined }) =>
undefined,
+ );
+
+ const extracted = extract();
+ const otherTaskResult = other();
+ transform({ extracted, otherTaskResult });
+
+ expect(getDagTaskInputs(dag).get("transform")).toEqual({ extracted,
otherTaskResult });
+ });
+
it("records frozen inputs that later mutation of the caller's object cannot
change", () => {
const dag = new Dag("example_dag");
- const extracted = dag.task("extract", async () => undefined);
+ const extract = dag.task("extract", async () => undefined);
+ const transform = dag.task("transform", async (_: { extracted: undefined
}) => undefined);
+
+ const extracted = extract();
const inputs: Record<string, TaskRef> = { extracted };
- dag.task("transform", async () => undefined, { inputs });
+ transform(inputs as { extracted: TaskRef });
inputs.sneaky = extracted;
- const recorded = getDagTaskRecords(dag).get("transform")!.inputs;
+ const recorded = getDagTaskInputs(dag).get("transform")!;
expect(recorded).toEqual({ extracted });
expect(Object.isFrozen(recorded)).toBe(true);
});
@@ -72,36 +123,237 @@ describe("Dag", () => {
it("rejects an input taken from another Dag", () => {
const first = new Dag("first_dag");
const second = new Dag("second_dag");
- const extracted = first.task("extract", async () => undefined);
- expect(() =>
- second.task("transform", async () => undefined, { inputs: { extracted }
}),
- ).toThrowError(
+ const extracted = first.task("extract", async () => undefined)();
+ const transform = second.task("transform", async (_: { extracted: TaskRef
}) => undefined);
+
+ expect(() => transform({ extracted })).toThrowError(
/Input "extracted" of task "transform" comes from Dag "first_dag", not
"second_dag"/,
);
});
+ it("rejects a reference from another Dag object carrying the same Dag ID",
() => {
+ const first = new Dag("same_id");
+ const second = new Dag("same_id");
+ first.task("extract", async () => undefined);
+ const extracted = second.task("extract", async () => undefined)();
+ const transform = first.task("transform", async (_: { extracted: TaskRef
}) => undefined);
+
+ expect(() => transform({ extracted })).toThrowError(
+ /Input "extracted" of task "transform" was not returned by this Dag's
"extract"/,
+ );
+ });
+
+ it("rejects a reference to a task this Dag never registered", () => {
+ const first = new Dag("same_id");
+ const second = new Dag("same_id");
+ const ghost = second.task("ghost", async () => undefined)();
+ const transform = first.task("transform", async (_: { ghost: TaskRef }) =>
undefined);
+
+ expect(() => transform({ ghost })).toThrowError(
+ /Input "ghost" of task "transform" refers to unregistered task "ghost"/,
+ );
+ });
+
+ it("rejects a reference buried inside a literal input", () => {
+ // It would draw no edge, so the task would run without the upstream it was
+ // given. TypeScript rejects it for a well-typed argument, which leaves the
+ // `any`, the cast and the plain-JavaScript caller to this check.
+ const dag = new Dag("nested_dag");
+ const extract = dag.task("extract", async () => 1);
+ const other = dag.task("other", async () => 2);
+ const fan = dag.task("fan", async (_: { sources: unknown }) => undefined);
+ const sources = [extract(), other()];
+
+ expect(() => fan({ sources } as unknown as { sources: never
})).toThrowError(
+ /Input "sources" of task "fan" holds a reference to "extract" inside a
literal value/,
+ );
+ });
+
+ it("finds a reference nested several levels down, and survives a
self-reference", () => {
+ const dag = new Dag("deep_dag");
+ const extract = dag.task("extract", async () => 1);
+ const fan = dag.task("fan", async (_: { config: unknown }) => undefined);
+ const config: Record<string, unknown> = { outer: { inner: [{ from:
extract() }] } };
+ config["self"] = config;
+
+ expect(() => fan({ config } as unknown as { config: never })).toThrowError(
+ /holds a reference to "extract" inside a literal value/,
+ );
+ });
+
+ it("rejects an input keyed by a symbol", () => {
+ const dag = new Dag("symbol_dag");
+ const transform = dag.task("transform", async (_: { real: string }) =>
undefined);
+ const inputs = { real: "ok", [Symbol("sneaky")]: "value" };
+
+ expect(() => transform(inputs)).toThrowError(
+ /Input "Symbol\(sneaky\)" of task "transform" is keyed by a symbol; an
argument name is a string/,
+ );
+ });
+
+ it("records a positional call in argument order", () => {
+ const dag = new Dag("example_dag");
+ const extract = dag.task("extract", async (): Promise<number> => 1);
+ const transform = dag.task("transform", async (rows: number, region:
string) => `${region}`);
+
+ const extracted = extract();
+ transform(extracted, "us");
+
+ expect(getDagTaskInputs(dag).get("transform")).toEqual({ arg0: extracted,
arg1: "us" });
+
expect(Object.keys(getDagTaskInputs(dag).get("transform")!)).toEqual(["arg0",
"arg1"]);
+ });
+
+ it("names positional arguments from argNames", () => {
+ const dag = new Dag("example_dag");
+ const extract = dag.task("extract", async (): Promise<number> => 1);
+ const transform = dag.task("transform", async (rows: number, region:
string) => `${region}`, {
+ argNames: ["rows", "region"],
+ });
+
+ const extracted = extract();
+ transform(extracted, "us");
+
+ expect(getDagTaskInputs(dag).get("transform")).toEqual({ rows: extracted,
region: "us" });
+ });
+
+ it("labels the arguments argNames does not reach", () => {
+ const dag = new Dag("example_dag");
+ const transform = dag.task("transform", async (rows: number, region:
string) => `${region}`, {
+ argNames: ["rows"],
+ });
+
+ transform(1, "us");
+
+ expect(getDagTaskInputs(dag).get("transform")).toEqual({ rows: 1, arg1:
"us" });
+ });
+
it.each([
- ["a plain string", "extract"],
- ["an object without a dagId", { taskId: "extract" }],
- ["null", null],
- ])("rejects an input that is not a task handle: %s", (_label, value) => {
+ ["not an array", { argNames: 1 }, /argNames for Dag "d" task "t" must be
an array of names/],
+ ["not a string", { argNames: [1] }, /holds 1; each name must be a
non-empty string/],
+ ["empty", { argNames: [""] }, /holds ""; each name must be a non-empty
string/],
+ ["a number", { argNames: ["0"] }, /holds "0"; each name must be a
non-empty string/],
+ ["a duplicate", { argNames: ["a", "a"] }, /argNames for Dag "d" task "t"
names "a" twice/],
+ ])("rejects argNames that are %s", (_label, options, expected) => {
+ const dag = new Dag("d");
+
+ expect(() => dag.task("t", async (a: number) => a, options as
never)).toThrowError(expected);
+ });
+
+ it("reads a single argument that is not a map of names as one positional
input", () => {
const dag = new Dag("example_dag");
- const extracted = value as unknown as TaskRef;
- expect(() =>
- dag.task("transform", async () => undefined, { inputs: { extracted } }),
- ).toThrowError(
- /Input "extracted" of task "transform" must be a task handle returned by
dag\.task\(\.\.\.\)/,
+ const when = new Date();
+ const transform = dag.task("transform", async (at: Date) => at);
+
+ transform(when as never);
+
+ expect(getDagTaskInputs(dag).get("transform")).toEqual({ arg0: when });
+ });
+
+ it("reads a single reference as one positional input rather than a map of
names", () => {
+ const dag = new Dag("example_dag");
+ const extract = dag.task("extract", async (): Promise<{ rows: number }> =>
({ rows: 1 }));
+ const load = dag.task("load", async (totals: { rows: number }) =>
totals.rows);
+
+ const extracted = extract();
+ load(extracted);
+
+ expect(getDagTaskInputs(dag).get("load")).toEqual({ arg0: extracted });
+ });
+
+ it("spreads a positional task's bound arguments back into its argument
list", async () => {
+ const dag = new Dag("example_dag");
+ const seen: unknown[] = [];
+ const transform = dag.task(
+ "transform",
+ async (rows: number, region: string) => {
+ seen.push(rows, region);
+ },
+ { argNames: ["rows", "region"] },
);
- expect(getDagTaskRecords(dag).has("transform")).toBe(false);
+
+ transform(1, "us");
+ // The order the runtime hands the bound arguments over in, which is the
+ // order they were recorded.
+ await new Bundle(dag).getTaskHandler("example_dag", "transform")!({
+ rows: 1,
+ region: "us",
+ } as never);
+
+ expect(seen).toEqual([1, "us"]);
});
- it("rejects an input referring to a task that is not registered yet", () => {
+ it("calls a task declaring one object of named arguments with that object",
async () => {
const dag = new Dag("example_dag");
- expect(() =>
- dag.task("transform", async () => undefined, {
- inputs: { ghost: { dagId: "example_dag", taskId: "ghost" } },
- }),
- ).toThrowError(/Input "ghost" of task "transform" refers to unregistered
task "ghost"/);
+ const seen: unknown[] = [];
+ const store = dag.task("store", async ({ rows }: { rows: number }) => {
+ seen.push(rows);
+ });
+
+ store({ rows: 1 });
+ await new Bundle(dag).getTaskHandler("example_dag", "store")!({ rows: 7 }
as never);
+
+ expect(seen).toEqual([7]);
+ });
+
+ it("rejects calling the same task twice", () => {
+ const dag = new Dag("example_dag");
+ const extract = dag.task("extract", async () => undefined);
+ extract();
+
+ expect(() => extract()).toThrowError(
+ /Task "extract" of Dag "example_dag" was already called; a task holds
one place in a Dag/,
+ );
+ });
+
+ it("fails when the Dag is read with a task that was never called", () => {
+ const dag = new Dag("unplaced_dag");
+ dag.task("extract", async () => undefined)();
+ dag.task("orphan", async () => undefined);
+
+ expect(() => finalizeDag(dag)).toThrowError(
+ /Task "orphan" of Dag "unplaced_dag" is never called, so it has no place
in the Dag/,
+ );
+ });
+
+ it("reports the same failure on a second read rather than reporting itself
as read", () => {
+ const dag = new Dag("unplaced_dag");
+ dag.task("orphan", async () => undefined);
+
+ expect(() => finalizeDag(dag)).toThrowError(/is never called/);
+ expect(() => finalizeDag(dag)).toThrowError(/is never called/);
+ });
+
+ it("surfaces an uncalled task when a bundle reports what it provides", () =>
{
+ const dag = new Dag("served_dag");
+ dag.task("orphan", async () => undefined);
+ const bundle = new Bundle(dag);
+
+ expect(() => finalizeBundleDags(bundle)).toThrowError(
+ /Task "orphan" of Dag "served_dag" is never called/,
+ );
+ });
+
+ it("rejects a task added after the Dag was read", () => {
+ const dag = new Dag("closed_dag");
+ dag.task("extract", async () => undefined)();
+ finalizeDag(dag);
+
+ expect(() => dag.task("late", async () => undefined)).toThrowError(
+ /Task "late" cannot be added to Dag "closed_dag" after the Dag was read/,
+ );
+ });
+
+ it("rejects wiring through a factory after the Dag was read", () => {
+ // A factory outlives the module that built it, so a stray later call has
to
+ // be rejected rather than silently rewiring a Dag Airflow already read.
+ const dag = new Dag("closed_dag");
+ const extract = dag.task("extract", async () => undefined);
+ extract();
+ finalizeDag(dag);
+
+ expect(() => extract()).toThrowError(
+ /Task "extract" of Dag "closed_dag" was called after the Dag was read/,
+ );
});
it("retains its spec and each task's handler and spec, copied and frozen",
() => {
@@ -109,7 +361,7 @@ describe("Dag", () => {
const taskSpec = {};
const handler = async () => "hello";
const dag = new Dag("example_dag", dagSpec);
- dag.task("my_task", handler, { spec: taskSpec });
+ dag.task("my_task", handler, { spec: taskSpec })();
expect(dag.dagId).toBe("example_dag");
expect(dag.spec).toEqual(dagSpec);
@@ -155,9 +407,9 @@ describe("Dag", () => {
});
it.each([
- ["a misspelled inputs key", { input: {} }],
+ ["inputs, which the factory call now carries", { inputs: {} }],
["a misspelled spec key", { specs: {} }],
- ["an upstream handle passed positionally", { upstream: { dagId: "d",
taskId: "t" } }],
+ ["an upstream reference passed as an option", { upstream: { dagId: "d",
taskId: "t" } }],
])("rejects %s in the task options", (_label, options) => {
const dag = new Dag("example_dag");
expect(() =>
@@ -169,7 +421,7 @@ describe("Dag", () => {
it.each([
["null", null],
["an array", []],
- ["a string", "inputs"],
+ ["a string", "spec"],
["a non-plain object", new Date()],
])("rejects task options that are not an options object: %s", (_label,
options) => {
const dag = new Dag("example_dag");
@@ -178,16 +430,6 @@ describe("Dag", () => {
).toThrowError(/options for Dag "example_dag" task "transform" must be an
object/);
});
- it("rejects task inputs that are not a plain object", () => {
- const dag = new Dag("example_dag");
- expect(() =>
- dag.task("transform", async () => undefined, {
- inputs: new Date() as unknown as Record<string, TaskRef>,
- }),
- ).toThrowError(/inputs for Dag "example_dag" task "transform" must be an
object/);
- expect(dag.taskIds).toEqual([]);
- });
-
it("rejects duplicate taskIds within a Dag", () => {
const dag = new Dag("example_dag");
dag.task("dup", async () => undefined);
@@ -199,8 +441,8 @@ describe("Dag", () => {
const second = async () => "second";
const firstDag = new Dag("first_dag");
const secondDag = new Dag("second_dag");
- firstDag.task("extract", first);
- secondDag.task("extract", second);
+ firstDag.task("extract", first)();
+ secondDag.task("extract", second)();
const bundle = new Bundle();
bundle.register(firstDag, secondDag);
@@ -211,7 +453,7 @@ describe("Dag", () => {
it("accepts a Unicode dagId that Python's word-character rule allows", () =>
{
const handler = async () => undefined;
const dag = new Dag("café_dag");
- dag.task("任務", handler);
+ dag.task("任務", handler)();
const bundle = new Bundle();
bundle.register(dag);
expect(bundle.getTaskHandler("café_dag", "任務")).toBe(handler);
@@ -226,7 +468,7 @@ describe("Dag", () => {
it("treats a dotted TaskGroup taskId as a single taskId (group.task)", () =>
{
const dag = new Dag("example_dag");
- dag.task("transforms.normalize", async () => "ok");
+ dag.task("transforms.normalize", async () => "ok")();
const bundle = new Bundle();
bundle.register(dag);
expect(bundle.getTaskHandler("example_dag",
"transforms.normalize")).toBeDefined();
diff --git a/ts-sdk/tests/sdk/task-handler.test.ts
b/ts-sdk/tests/sdk/task-handler.test.ts
index 78bb1541802..bf406321f0e 100644
--- a/ts-sdk/tests/sdk/task-handler.test.ts
+++ b/ts-sdk/tests/sdk/task-handler.test.ts
@@ -19,7 +19,7 @@
import { describe, expect, it } from "vitest";
-import { Bundle, listBundleDags, listBundleTasks } from
"../../src/sdk/bundle.js";
+import { Bundle, bundleDagTaskIds } from "../../src/sdk/bundle.js";
import { Dag } from "../../src/sdk/dag.js";
import { getTaskHandlerFunction, TaskHandler } from
"../../src/sdk/task-handler.js";
@@ -66,7 +66,7 @@ describe("TaskHandler", () => {
describe("a bundle of task handlers", () => {
it("registers Dags and task handlers in one call", () => {
const nativeDag = new Dag("native_etl");
- nativeDag.task("extract", async () => undefined);
+ nativeDag.task("extract", async () => undefined)();
const transform = async () => "transformed";
const bundle = new Bundle();
@@ -94,9 +94,9 @@ describe("a bundle of task handlers", () => {
expect(bundle.getTaskHandler("etl", "unknown")).toBeUndefined();
});
- it("lists every Dag it provides for, in registration order", () => {
+ it("lists every Dag it provides for, the ones declared in TypeScript first",
() => {
const nativeDag = new Dag("native_etl");
- nativeDag.task("extract", async () => undefined);
+ nativeDag.task("extract", async () => undefined)();
const bundle = new Bundle(
new TaskHandler("py_etl", "transform", async () => undefined),
@@ -104,16 +104,12 @@ describe("a bundle of task handlers", () => {
new TaskHandler("py_etl", "report", async () => undefined),
);
- // A Dag registered through handlers keeps its place from the first handler
- // that named it, so a later one does not reorder the manifest.
- expect(listBundleDags(bundle)).toEqual([
- { dagId: "py_etl", tasks: ["transform", "report"] },
- { dagId: "native_etl", tasks: ["extract"] },
- ]);
- expect(listBundleTasks(bundle)).toEqual([
- { dagId: "py_etl", taskId: "transform" },
- { dagId: "py_etl", taskId: "report" },
- { dagId: "native_etl", taskId: "extract" },
+ // Compared as entries rather than as a Map, which would ignore the order.
+ // A Dag its handlers name keeps its place from the first handler that
named
+ // it, so a later one does not reorder the manifest.
+ expect([...bundleDagTaskIds(bundle)]).toEqual([
+ ["native_etl", ["extract"]],
+ ["py_etl", ["transform", "report"]],
]);
});
@@ -122,7 +118,7 @@ describe("a bundle of task handlers", () => {
bundle.register(new TaskHandler("py_etl", "transform", async () =>
undefined));
bundle.register(new TaskHandler("py_etl", "report", async () =>
undefined));
- expect(listBundleDags(bundle)).toEqual([{ dagId: "py_etl", tasks:
["transform", "report"] }]);
+ expect(bundleDagTaskIds(bundle)).toEqual(new Map([["py_etl", ["transform",
"report"]]]));
});
it("rejects a second handler for the same Dag and task", () => {
@@ -150,7 +146,7 @@ describe("a bundle of task handlers", () => {
new TaskHandler("etl", "transform", async () => undefined),
),
).toThrowError(/already registered/);
- expect(listBundleDags(bundle)).toEqual([]);
+ expect(bundleDagTaskIds(bundle)).toEqual(new Map());
expect(bundle.getTaskHandler("etl", "transform")).toBeUndefined();
});
@@ -180,6 +176,42 @@ describe("a bundle of task handlers", () => {
);
});
+ it("rejects a Dag whose ID the same call already gave a handler", () => {
+ // The mirror of the Dag-first case, which the validation pass has to catch
+ // too: writing the Dag entry would otherwise replace the handler map and
+ // drop the handler, leaving its task with no body at run time.
+ const dag = new Dag("py_etl");
+ dag.task("extract", async () => undefined)();
+ expect(
+ () => new Bundle(new TaskHandler("py_etl", "legacy", async () =>
undefined), dag),
+ ).toThrowError(/already has registered task handlers/);
+ });
+
+ it("registers neither kind when a mixed call throws", () => {
+ const bundle = new Bundle();
+ const dag = new Dag("py_etl");
+ dag.task("extract", async () => undefined)();
+ expect(() =>
+ bundle.register(new TaskHandler("py_etl", "legacy", async () =>
undefined), dag),
+ ).toThrowError(/already has registered task handlers/);
+
+ expect(bundleDagTaskIds(bundle)).toEqual(new Map());
+ expect(bundle.getTaskHandler("py_etl", "legacy")).toBeUndefined();
+ });
+
+ it("keeps the handlers it already held when a later call throws", () => {
+ const bundle = new Bundle(new TaskHandler("py_etl", "transform", async ()
=> undefined));
+ expect(() =>
+ bundle.register(
+ new TaskHandler("py_etl", "report", async () => undefined),
+ new TaskHandler("py_etl", "report", async () => undefined),
+ ),
+ ).toThrowError(/already registered/);
+
+ expect(bundleDagTaskIds(bundle)).toEqual(new Map([["py_etl",
["transform"]]]));
+ expect(bundle.getTaskHandler("py_etl", "report")).toBeUndefined();
+ });
+
it("names the duplicate-copy cause for a handler carrying the brand but not
this class", () => {
// Stands in for a handler from a second resolved copy: same brand, other
// class. Its private function field is unreadable here, so the point is
