This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 65a67bb411f [v3-4-test] TS SDK: Reject blank edge labels and label
task edges into groups (#75190) (#75385)
65a67bb411f is described below
commit 65a67bb411f9ceef7fc027013e0aba93c4e3a072
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Oct 22 23:47:32 2026 +0530
[v3-4-test] TS SDK: Reject blank edge labels and label task edges into
groups (#75190) (#75385)
* TS SDK: Reject blank edge labels and label task edges into groups
* Address review comments on edge labels
---------
(cherry picked from commit c558f4a60f3a5d34e3d9903539b8536629ec41c3)
Signed-off-by: PoAn Yang <[email protected]>
Co-authored-by: PoAn Yang <[email protected]>
---
.../language-sdks/typescript.rst | 24 ++-
ts-sdk/api-docs/dag-authoring-api.ts | 2 +-
.../ci/prek/check_serialization_conformance.py | 2 +-
ts-sdk/src/coordinator/serde.ts | 36 +++-
ts-sdk/src/index.ts | 2 +-
ts-sdk/src/sdk/dag-graph.ts | 19 +-
ts-sdk/src/sdk/dag.ts | 209 ++++++++++++++++----
ts-sdk/tests/conformance/serialize_typescript.ts | 7 +-
ts-sdk/tests/coordinator/serde.test.ts | 220 ++++++++++++++++++++-
ts-sdk/tests/public-api.test.ts | 10 +-
ts-sdk/tests/sdk/branching.test.ts | 8 +-
ts-sdk/tests/sdk/conditional.test.ts | 10 +-
ts-sdk/tests/sdk/dag.test.ts | 133 ++++++++++---
13 files changed, 577 insertions(+), 105 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 5b2ca7a816f..b9937aa8829 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -243,23 +243,33 @@ the reference it was called on, so
``loaded.before(cleaned).before(notified)`` d
Pass a value as an input when the downstream task needs it, and use ``before``
or ``after`` when it only needs
to run in order.
-Wrap a reference in ``label`` to name the edge drawn to it, as Python's
``Label`` does:
+To label an edge the way Python's ``Label`` does, pass
``Label.of(text).on(reference)`` to ``before`` or
+``after``. The text comes before the reference. That way the text sits between
the two tasks of the edge in
+the code, just as the label sits between them in the graph:
.. code-block:: typescript
- import { label } from "apache-airflow-ts-sdk";
+ import { Label } from "apache-airflow-ts-sdk";
- checked.before(label(processed, "rows found"), label(notified, "no rows"));
+ checked.before(Label.of("rows found").on(processed), Label.of("no
rows").on(notified));
+ cleaned.after(Label.of("always").on(loaded)); // loaded >>
Label("always") >> cleaned
-A labeled reference only goes into ``before`` or ``after``; calling either on
it throws. Redrawing an
-existing edge, including one an input drew, labels it. That is also how to
label the branches of ``dag.if``
-or ``dag.switch`` (see below):
+``on`` also takes several references and puts the same label on the edge to or
from each of them:
+
+.. code-block:: typescript
+
+ // loaded >> Label("forks") >> [reportedEmpty, notified]
+ loaded.before(Label.of("forks").on(reportedEmpty, notified));
+
+Pass the value that ``on`` returns only to ``before`` or ``after``. Calling
``before`` or ``after`` on that
+value throws an error. Redrawing an existing edge, including one an input
drew, labels it. That is also how
+to label the branches of ``dag.if`` or ``dag.switch`` (see below):
.. code-block:: typescript
const rowCheck = dag.if(hasRows, { total });
rowCheck.then(loaded).else(reportedEmpty);
- rowCheck.before(label(loaded, "rows found"), label(reportedEmpty, "no
rows"));
+ rowCheck.before(Label.of("rows found").on(loaded), Label.of("no
rows").on(reportedEmpty));
Task groups
~~~~~~~~~~~
diff --git a/ts-sdk/api-docs/dag-authoring-api.ts
b/ts-sdk/api-docs/dag-authoring-api.ts
index 8fd6cc4427c..74b53306244 100644
--- a/ts-sdk/api-docs/dag-authoring-api.ts
+++ b/ts-sdk/api-docs/dag-authoring-api.ts
@@ -24,7 +24,7 @@ export {
Dag,
getClient,
getContext,
- label,
+ Label,
NEVER_EXPIRE,
TaskHandler,
triggerDagRun,
diff --git a/ts-sdk/scripts/ci/prek/check_serialization_conformance.py
b/ts-sdk/scripts/ci/prek/check_serialization_conformance.py
index ceaa0928959..95a11280f2f 100755
--- a/ts-sdk/scripts/ci/prek/check_serialization_conformance.py
+++ b/ts-sdk/scripts/ci/prek/check_serialization_conformance.py
@@ -29,7 +29,7 @@ from common_prek_utils import AIRFLOW_ROOT_PATH, run_command
# The features of test_dags.yaml that
tests/conformance/serialize_typescript.ts can build. A Dag that
# requires another feature is left out until the builder gets it.
-SUPPORTED_FEATURES: list[str] = []
+SUPPORTED_FEATURES: list[str] = ["edge_labels"]
if __name__ not in ("__main__", "__mp_main__"):
raise SystemExit(
diff --git a/ts-sdk/src/coordinator/serde.ts b/ts-sdk/src/coordinator/serde.ts
index ba789f0c907..c528980487c 100644
--- a/ts-sdk/src/coordinator/serde.ts
+++ b/ts-sdk/src/coordinator/serde.ts
@@ -42,7 +42,7 @@ import {
import type { JsonValue } from "../sdk/client-types.js";
import {
DOWNSTREAM_JOIN_SUFFIX,
- getDagOrderEdges,
+ getDagEdgeLabellings,
getDagTaskGroups,
getDagTaskInputs,
getDagTaskRecords,
@@ -87,15 +87,37 @@ const TASK_MODULE = "airflow.sdk.coordinators.node";
*/
const TASK_LANGUAGE = "typescript";
-/** Python's `DAG.edge_info`, keying a group end by its join node as Python
does. */
-function serializeEdgeInfo(dag: Dag): SerializedValue {
+/**
+ * Python's `DAG.edge_info`, keying a group end by its join node as Python
does.
+ *
+ * This replays all labels in the order they were set, because Python writes a
label each time it
+ * draws an edge. When two labels land on one entry, the label set last is the
one kept.
+ */
+function serializeEdgeInfo(dag: Dag, graph: DagGraph): SerializedValue {
const groups = getDagTaskGroups(dag);
+ const grouped = new Set([...groups.values()].flatMap((group) =>
group.taskIds));
const info: Record<string, Record<string, SerializedValue>> = {};
- for (const { upstream, downstream, label } of getDagOrderEdges(dag)) {
- if (label === undefined) continue;
+ const put = (from: string, to: string, label: string): void => {
+ (info[from] ??= {})[to] = { label };
+ };
+ for (const { upstream, downstream, label, receiver } of
getDagEdgeLabellings(dag)) {
const from = groups.has(upstream) ? upstream + DOWNSTREAM_JOIN_SUFFIX :
upstream;
const to = groups.has(downstream) ? downstream + UPSTREAM_JOIN_SUFFIX :
downstream;
- (info[from] ??= {})[to] = { label };
+ put(from, to, label);
+ // Python's `extract >> Label("x") >> group` also labels the edge from
extract to each root of
+ // the group. Python's `report << Label("x") << group` labels the edge
from each leaf of the
+ // group to report. Python labels these task edges only when a task is the
receiver, as in
+ // `extracted.before(...)` or `reported.after(...)`, and only when no
group holds that task.
+ // When a group holds the task, Python replaces the task with that group,
and the label goes
+ // on the edge between the two groups. This code keeps the task as the end
of the edge,
+ // because a label never changes the ends of its edge. Neither Python nor
this code labels a
+ // task edge in that case.
+ const task = receiver === "upstream" ? upstream : downstream;
+ if (groups.has(task) || grouped.has(task)) continue;
+ for (const end of graph.groupEndsReached.get(upstream)?.get(downstream) ??
[]) {
+ if (receiver === "upstream") put(upstream, end, label);
+ else put(end, downstream, label);
+ }
}
return info;
}
@@ -186,7 +208,7 @@ export function serializeDag(
}),
dag_dependencies: dagDependencies,
task_group: serializeTaskGroups(dag, graph),
- edge_info: serializeEdgeInfo(dag),
+ edge_info: serializeEdgeInfo(dag, graph),
params: [],
// Always written by Python's serializer, so a Dag without either still
// round-trips to the same object.
diff --git a/ts-sdk/src/index.ts b/ts-sdk/src/index.ts
index 865eae75df2..8cad4298fb6 100644
--- a/ts-sdk/src/index.ts
+++ b/ts-sdk/src/index.ts
@@ -17,7 +17,7 @@
* under the License.
*/
-export { Dag, label } from "./sdk/dag.js";
+export { Dag, Label } from "./sdk/dag.js";
export { triggerDagRun } from "./sdk/trigger-dag-run.js";
export { Bundle } from "./sdk/bundle.js";
export { TaskHandler } from "./sdk/task-handler.js";
diff --git a/ts-sdk/src/sdk/dag-graph.ts b/ts-sdk/src/sdk/dag-graph.ts
index 4d7da55902b..b07b524b228 100644
--- a/ts-sdk/src/sdk/dag-graph.ts
+++ b/ts-sdk/src/sdk/dag-graph.ts
@@ -49,6 +49,15 @@ export interface DagGraph {
/** Each task's downstream task IDs, with every group endpoint expanded. */
readonly downstreamTaskIds: Map<string, Set<string>>;
readonly groupEdges: Map<string, GroupEdgeSets>;
+ /**
+ * For each labelled edge between a task and a group, the tasks at the
group's end of the edge:
+ * the group's roots when the group is downstream, its leaves when the group
is upstream. Keyed
+ * by the upstream ID and then by the downstream ID. The roots and leaves
are the ones the group
+ * has when that edge is expanded. To find them, every edge between two
tasks counts, whenever
+ * it was drawn. An edge to or from a group counts only if it was drawn
before the labelled
+ * edge.
+ */
+ readonly groupEndsReached: Map<string, Map<string, readonly string[]>>;
}
/**
@@ -65,6 +74,7 @@ export function buildDagGraph(dag: Dag): DagGraph {
const groups = getDagTaskGroups(dag);
const downstreamTaskIds = new Map<string, Set<string>>();
const groupEdges = new Map<string, GroupEdgeSets>();
+ const groupEndsReached = new Map<string, Map<string, readonly string[]>>();
const link = (upstream: string, downstream: string): void => {
// Two arguments fed by the same upstream are one edge, as is an order-only
// edge redeclaring one the wiring already drew.
@@ -149,13 +159,18 @@ export function buildDagGraph(dag: Dag): DagGraph {
return sets;
};
- for (const { upstream, downstream } of orderEdges) {
+ for (const { upstream, downstream, label } of orderEdges) {
const upstreamIsGroup = groups.has(upstream);
const downstreamIsGroup = groups.has(downstream);
if (!upstreamIsGroup && !downstreamIsGroup) continue;
const from = tasksAt(upstream, "upstream");
const to = tasksAt(downstream, "downstream");
+ if (upstreamIsGroup !== downstreamIsGroup && label !== undefined) {
+ const reached = groupEndsReached.get(upstream) ?? new Map<string,
readonly string[]>();
+ reached.set(downstream, upstreamIsGroup ? ends.leaves(upstream) :
ends.roots(downstream));
+ groupEndsReached.set(upstream, reached);
+ }
for (const tail of from) {
for (const head of to) link(tail, head);
}
@@ -174,7 +189,7 @@ export function buildDagGraph(dag: Dag): DagGraph {
else sets.downstreamTasks.add(downstream);
}
}
- return { downstreamTaskIds, groupEdges };
+ return { downstreamTaskIds, groupEdges, groupEndsReached };
}
function addTo(index: Map<string, Set<string>>, key: string, value: string):
void {
diff --git a/ts-sdk/src/sdk/dag.ts b/ts-sdk/src/sdk/dag.ts
index 398d5fc7fb3..bffb4e77b2b 100644
--- a/ts-sdk/src/sdk/dag.ts
+++ b/ts-sdk/src/sdk/dag.ts
@@ -273,6 +273,17 @@ export interface OrderEdge {
readonly label?: string;
}
+/**
+ * Internal: one label that a call of `before` or `after` set on an order-only
edge. `receiver` is
+ * the end of the edge whose method was called.
+ */
+export interface EdgeLabelling {
+ readonly upstream: string;
+ readonly downstream: string;
+ readonly label: string;
+ readonly receiver: "upstream" | "downstream";
+}
+
// A task id cannot hold a NUL, so a joined pair cannot collide with one.
const EDGE_KEY_SEPARATOR = "\u0000";
@@ -327,48 +338,113 @@ export function isTaskRef(value: unknown): value is
TaskRef {
const conditionTasks = new WeakMap<object, TaskRef>();
-const labelledNodes = new WeakMap<object, { readonly node: Node; readonly
text: string }>();
+const labelledNodes = new WeakMap<
+ object,
+ { readonly nodes: readonly Node[]; readonly text: string }
+>();
function resolveNode(node: Node): Node {
if (typeof node !== "object" || node === null) return node;
- return labelledNodes.get(node)?.node ?? conditionTasks.get(node) ?? node;
+ return conditionTasks.get(node) ?? node;
}
/**
- * Label the edge that `before` or `after` draws to `node`, as Python's
`Label` does.
- * The result only goes into `before` or `after`; calling either on it throws.
- * `text` cannot be empty.
+ * A text label for an edge. The Airflow UI shows the label on the edge in the
graph, as it does
+ * for Python's `Label`. The text comes before the task that the label is put
on. That way the
+ * text sits between the two tasks of the edge in the code, just as the label
sits between them in
+ * the graph:
*
* ```ts
- * checked.before(label(processed, "rows found"), label(emptyNotice, "no
rows"));
+ * // load >> Label("when empty") >> report_empty
+ * loaded.before(Label.of("when empty").on(reportedEmpty));
+ * // load >> Label("always") >> cleanup
+ * cleaned.after(Label.of("always").on(loaded));
* ```
*/
-export function label(node: Node, text: string): Node {
- if (typeof text !== "string") {
- throw new Error(`label() takes a string to name the edge with, not
${describeValue(text)}`);
+export class Label {
+ readonly #text: string;
+
+ // The constructor is private only in TypeScript. Plain JavaScript can still
call
+ // `new Label("")`, so the text check lives here and not in `of`. A check in
`of` would not run
+ // for that call.
+ private constructor(text: string) {
+ if (typeof text !== "string") {
+ throw new Error(
+ `Label.of() takes a string to name the edge with, not
${describeValue(text)}`,
+ );
+ }
+ if (text.trim() === "") {
+ throw new Error(
+ "Label.of() takes the text to put on the edge, not " +
+ (text === "" ? "an empty string" : "a blank string"),
+ );
+ }
+ this.#text = text;
}
- if (text === "") {
- throw new Error("label() takes the text to put on the edge, not an empty
string");
+
+ /**
+ * Make a label with the given `text`. Use `on` to put the label on edges.
`text` must not be
+ * empty or only whitespace.
+ */
+ static of(text: string): Label {
+ return new Label(text);
}
- const target = resolveNode(node);
- const id = nodeId(target);
- if (id === undefined) {
- throw new Error(`label() takes a task or task group reference, not
${describeValue(node)}`);
+
+ /**
+ * Put the label on each edge that a `before` or `after` call draws between
its receiver and
+ * `node` or a node in `more`.
+ *
+ * So each edge of a fan-out can have its own label, and one label can go on
several edges:
+ *
+ * ```ts
+ * checked.before(Label.of("rows found").on(processed), Label.of("no
rows").on(reportedEmpty));
+ * // load >> Label("forks") >> [report_empty, notify]
+ * loaded.before(Label.of("forks").on(reportedEmpty, notified));
+ * ```
+ *
+ * Pass the value that `on` returns to `before` or `after`. Calling `before`
or `after` on that
+ * value throws an error.
+ */
+ on(node: Node, ...more: readonly Node[]): Node {
+ const text = this.#text;
+ const nodes = [node, ...more].flatMap((each) => {
+ // Labelling a labelled node again replaces its label.
+ const labelled = labelledNodes.get(each);
+ if (labelled !== undefined) return labelled.nodes;
+ const target = resolveNode(each);
+ if (nodeId(target) === undefined) {
+ throw new Error(
+ `Label.of(${JSON.stringify(text)}).on() takes a task or task group
reference, ` +
+ `not ${describeValue(each)}`,
+ );
+ }
+ return [target];
+ });
+ const { dagId } = nodes[0]!;
+ const stranger = nodes.find((each) => each.dagId !== dagId);
+ if (stranger !== undefined) {
+ throw new Error(
+ `Label.of(${JSON.stringify(text)}).on() takes nodes of one Dag, but ` +
+ `"${nodeId(nodes[0]!)}" is in Dag "${dagId}" and
"${nodeId(stranger)}" is in Dag ` +
+ `"${stranger.dagId}"`,
+ );
+ }
+ const ids = nodes.map((each) => `"${nodeId(each)}"`).join(", ");
+ const misuse = (verb: "before" | "after"): never => {
+ throw new Error(
+ `Label.of(${JSON.stringify(text)}) on ${ids} of Dag "${dagId}" labels
edges, so pass it ` +
+ `to before() or after() rather than calling ${verb}() on it`,
+ );
+ };
+ const labelled: Node = {
+ dagId,
+ before: () => misuse("before"),
+ after: () => misuse("after"),
+ };
+ brand(labelled, "LabelledNode");
+ labelledNodes.set(labelled, { nodes, text });
+ return Object.freeze(labelled);
}
- const misuse = (verb: "before" | "after"): never => {
- throw new Error(
- `label(..., ${JSON.stringify(text)}) on "${id}" of Dag "${target.dagId}"
names the edge ` +
- `drawn to it, so pass it to before() or after() rather than calling
${verb}() on it`,
- );
- };
- const labelled: Node = {
- dagId: target.dagId,
- before: () => misuse("before"),
- after: () => misuse("after"),
- };
- brand(labelled, "LabelledNode");
- labelledNodes.set(labelled, { node: target, text });
- return Object.freeze(labelled);
}
function isLabelledNode(value: unknown): boolean {
@@ -388,8 +464,8 @@ function isLabelledNode(value: unknown): boolean {
function findNestedTaskRef(value: unknown, seen: WeakSet<object>): TaskRef |
undefined {
if (typeof value !== "object" || value === null) return undefined;
if (isTaskRef(value)) return value;
- const labelled = labelledNodes.get(value)?.node;
- if (labelled !== undefined) return isTaskRef(labelled) ? labelled :
undefined;
+ const labelled = labelledNodes.get(value);
+ if (labelled !== undefined) return labelled.nodes.find(isTaskRef);
// 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;
@@ -570,6 +646,7 @@ export type RecordedInputs = Readonly<Record<string,
TaskRef | JsonValue>>;
let taskRecordsOf: (dag: Dag) => ReadonlyMap<string, TaskRecord>;
let inputsOf: (dag: Dag) => ReadonlyMap<string, RecordedInputs>;
let orderEdgesOf: (dag: Dag) => readonly OrderEdge[];
+let labellingsOf: (dag: Dag) => readonly EdgeLabelling[];
let definedInOf: (dag: Dag) => string | undefined;
let groupsOf: (dag: Dag) => ReadonlyMap<string, TaskGroupRecord>;
let finalizeOf: (dag: Dag) => void;
@@ -608,6 +685,10 @@ export class Dag {
// Keyed by the two task ids, so declaring an edge twice records it once, and
// insertion-ordered so the serialized Dag reads as written.
readonly #orderEdges = new Map<string, OrderEdge>();
+ // Every label set on an order-only edge, in the order it was set. The
serializer replays them
+ // the way Python writes `edge_info`. The end that set a label decides which
task edges the
+ // label also covers, and the label set last on an entry is the one kept.
+ readonly #labellings: EdgeLabelling[] = [];
readonly #definedIn: string | undefined;
readonly #conditions = new Map<string, ConditionRecord>();
readonly #branches = new Map<string, readonly TaskRef[]>();
@@ -623,6 +704,7 @@ export class Dag {
taskRecordsOf = (dag) => dag.#tasks;
inputsOf = (dag) => dag.#inputs;
orderEdgesOf = (dag) => [...dag.#orderEdges.values()];
+ labellingsOf = (dag) => [...dag.#labellings];
definedInOf = (dag) => dag.#definedIn;
groupsOf = (dag) => dag.#groups;
finalizeOf = (dag) => dag.#finalize();
@@ -1189,7 +1271,27 @@ export class Dag {
}
#addOrderEdge(upstreamNode: Node, downstreamNode: Node, verb: "before" |
"after"): void {
- const text = labelledNodes.get(verb === "before" ? downstreamNode :
upstreamNode)?.text;
+ // Calling before() or after() on a labelled node throws, so only the
argument can be a
+ // labelled node.
+ const argument = verb === "before" ? downstreamNode : upstreamNode;
+ const labelled = labelledNodes.get(argument);
+ // Every edge is checked before the first one is recorded, so a label on
several nodes draws
+ // all of its edges or none of them.
+ const edges = (labelled?.nodes ?? [argument]).map((node) =>
+ verb === "before"
+ ? this.#checkOrderEdge(upstreamNode, node, verb)
+ : this.#checkOrderEdge(node, downstreamNode, verb),
+ );
+ for (const [upstreamId, downstreamId] of edges) {
+ this.#recordOrderEdge(upstreamId, downstreamId, labelled?.text, verb);
+ }
+ }
+
+ #checkOrderEdge(
+ upstreamNode: Node,
+ downstreamNode: Node,
+ verb: "before" | "after",
+ ): readonly [string, string] {
const upstream = resolveNode(upstreamNode);
const downstream = resolveNode(downstreamNode);
if (this.#finalized) {
@@ -1202,32 +1304,51 @@ export class Dag {
// Dag handed out, since it is what carries the method.
const other = verb === "before" ? downstream : upstream;
this.#validateOwnNode(other, `${verb}()`);
- const upstreamId = nodeId(upstream);
- const downstreamId = nodeId(downstream);
+ const upstreamId = nodeId(upstream)!;
+ const downstreamId = nodeId(downstream)!;
if (upstreamId === downstreamId) {
throw new Error(
`${verb}() cannot draw an edge from node "${upstreamId}" of Dag
"${this.dagId}" to ` +
"itself; an edge orders two different nodes",
);
}
+ return [upstreamId, downstreamId];
+ }
+
+ #recordOrderEdge(
+ upstreamId: string,
+ downstreamId: string,
+ text: string | undefined,
+ verb: "before" | "after",
+ ): void {
const key = `${upstreamId}${EDGE_KEY_SEPARATOR}${downstreamId}`;
// Idempotent, so an edge drawn from both ends is one edge.
const edgeLabel = text ?? this.#orderEdges.get(key)?.label;
this.#orderEdges.set(
key,
Object.freeze({
- upstream: upstreamId!,
- downstream: downstreamId!,
+ upstream: upstreamId,
+ downstream: downstreamId,
...(edgeLabel === undefined ? {} : { label: edgeLabel }),
}),
);
+ if (text !== undefined) {
+ this.#labellings.push(
+ Object.freeze({
+ upstream: upstreamId,
+ downstream: downstreamId,
+ label: text,
+ receiver: verb === "before" ? "upstream" : "downstream",
+ }),
+ );
+ }
}
#rejectLabel(node: unknown, context: string, deciderId: string): void {
if (isLabelledNode(node)) {
throw new Error(
- `${context} of Dag "${this.dagId}" was given a label(); label a branch
edge by ` +
- `redrawing it from "${deciderId}" with before(label(task, "text"))`,
+ `${context} of Dag "${this.dagId}" was given Label.of(...).on(...);
label a branch edge ` +
+ `by redrawing it from "${deciderId}" with
before(Label.of("text").on(task))`,
);
}
}
@@ -1236,7 +1357,8 @@ export class Dag {
const id = nodeId(node);
if (id === undefined && isLabelledNode(node)) {
throw new Error(
- `${context} on Dag "${this.dagId}" was given a label() that
${DUPLICATE_COPY_HINT}`,
+ `${context} on Dag "${this.dagId}" was given Label.of(...).on(...),
which ` +
+ DUPLICATE_COPY_HINT,
);
}
if (id === undefined) {
@@ -1305,8 +1427,8 @@ export class Dag {
const value = inputs[key];
if (isLabelledNode(value)) {
throw new Error(
- `Input "${key}" of task "${taskId}" was given a label(), which only
goes into ` +
- "before() or after(); pass the reference itself",
+ `Input "${key}" of task "${taskId}" was given Label.of(...).on(...),
which only goes ` +
+ "into before() or after(); pass the reference itself",
);
}
// Anything unbranded is a literal argument, including a look-alike
@@ -1440,6 +1562,11 @@ export function getDagOrderEdges(dag: Dag): readonly
OrderEdge[] {
return orderEdgesOf(dag);
}
+/** Internal: every label set on an order-only edge of a Dag, in the order it
was set. */
+export function getDagEdgeLabellings(dag: Dag): readonly EdgeLabelling[] {
+ return labellingsOf(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> {
diff --git a/ts-sdk/tests/conformance/serialize_typescript.ts
b/ts-sdk/tests/conformance/serialize_typescript.ts
index 948d12a1688..9dabc16a71c 100644
--- a/ts-sdk/tests/conformance/serialize_typescript.ts
+++ b/ts-sdk/tests/conformance/serialize_typescript.ts
@@ -37,6 +37,7 @@ import {
} from "../../src/generated/dag-schema-fields.js";
import {
Dag,
+ Label,
type DagSpec,
type TaskGroupRef,
type TaskRef,
@@ -55,7 +56,7 @@ interface DagCase {
readonly spec?: Readonly<Record<string, unknown>>;
readonly groups?: readonly string[];
readonly tasks: readonly TaskCase[];
- readonly order_edges?: readonly (readonly [string, string])[];
+ readonly order_edges?: readonly (readonly [string, string, string?])[];
}
// A duration is already a number of seconds in this SDK's authoring API.
@@ -132,12 +133,12 @@ function buildDag(dagCase: DagCase): Dag {
}
refs.set(qualifiedTaskId(task),
factories.get(qualifiedTaskId(task))!(inputs));
}
- for (const [upstream, downstream] of dagCase.order_edges ?? []) {
+ for (const [upstream, downstream, text] of dagCase.order_edges ?? []) {
const from = groups.get(upstream) ?? refs.get(upstream);
const to = groups.get(downstream) ?? refs.get(downstream);
if (!from || !to)
throw new Error(`${dagCase.dag_id}: no node "${from ? downstream :
upstream}"`);
- from.before(to);
+ from.before(text === undefined ? to : Label.of(text).on(to));
}
return dag;
}
diff --git a/ts-sdk/tests/coordinator/serde.test.ts
b/ts-sdk/tests/coordinator/serde.test.ts
index 0b15d715750..df508ec2871 100644
--- a/ts-sdk/tests/coordinator/serde.test.ts
+++ b/ts-sdk/tests/coordinator/serde.test.ts
@@ -27,7 +27,7 @@ import {
} from "../../src/coordinator/serde.js";
import {
Dag,
- label,
+ Label,
type DagSpec,
type TaskGroupRef,
type TaskRef,
@@ -37,6 +37,12 @@ import {
type Json = Record<string, unknown>;
type WiredArgs = Record<string, TaskRef>;
+interface StagingRefs {
+ readonly extracted: TaskRef;
+ readonly staging: TaskGroupRef;
+ readonly reported: TaskRef;
+}
+
/** Declare a task and place it, optionally behind some upstreams. */
function place(
scope: Dag | TaskGroupRef,
@@ -493,17 +499,221 @@ describe("serializeDag", () => {
const staging = dag.taskGroup("staging");
place(staging, "stage");
const loaded = place(dag, "load");
- extracted.before(label(transformed, "rows"));
- transformed.before(label(staging, "staged"));
- staging.before(label(loaded, "ready"));
+ extracted.before(Label.of("rows").on(transformed));
+ transformed.before(Label.of("staged").on(staging));
+ staging.before(Label.of("ready").on(loaded));
expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
extract: { transform: { label: "rows" } },
- transform: { "staging.upstream_join_id": { label: "staged" } },
+ transform: {
+ "staging.upstream_join_id": { label: "staged" },
+ "staging.stage": { label: "staged" },
+ },
"staging.downstream_join_id": { load: { label: "ready" } },
});
});
+ it.each([
+ [
+ "a task's before() labels the edges to the group's roots",
+ (refs: StagingRefs) =>
refs.extracted.before(Label.of("L").on(refs.staging)),
+ {
+ extract: {
+ "staging.upstream_join_id": { label: "L" },
+ "staging.stage": { label: "L" },
+ "staging.audit": { label: "L" },
+ },
+ },
+ ],
+ [
+ "a task's after() labels the edges from the group's leaves",
+ (refs: StagingRefs) =>
refs.reported.after(Label.of("L").on(refs.staging)),
+ {
+ "staging.downstream_join_id": { report: { label: "L" } },
+ "staging.check": { report: { label: "L" } },
+ "staging.audit": { report: { label: "L" } },
+ },
+ ],
+ [
+ "a group's after() labels only the edge to its join node",
+ (refs: StagingRefs) =>
refs.staging.after(Label.of("L").on(refs.extracted)),
+ { extract: { "staging.upstream_join_id": { label: "L" } } },
+ ],
+ [
+ "a group's before() labels only the edge from its join node",
+ (refs: StagingRefs) =>
refs.staging.before(Label.of("L").on(refs.reported)),
+ { "staging.downstream_join_id": { report: { label: "L" } } },
+ ],
+ ])("%s", (_name, draw, expected) => {
+ const dag = new Dag("d");
+ const extracted = place(dag, "extract");
+ const staging = dag.taskGroup("staging");
+ const staged = place(staging, "stage");
+ place(staging, "check", [staged]);
+ place(staging, "audit");
+ const reported = place(dag, "report");
+ draw({ extracted, staging, reported });
+
+ expect((serializeDag(dag, "", ".") as
Json)["edge_info"]).toEqual(expected);
+ });
+
+ it("keeps the root labels of before() when the group's after() relabels
the edge", () => {
+ const dag = new Dag("d");
+ const extracted = place(dag, "extract");
+ const staging = dag.taskGroup("staging");
+ place(staging, "stage");
+ extracted.before(Label.of("first").on(staging));
+ staging.after(Label.of("second").on(extracted));
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ extract: {
+ "staging.stage": { label: "first" },
+ "staging.upstream_join_id": { label: "second" },
+ },
+ });
+ });
+
+ it("labels no task edge of a group edge from a task in a group, or past an
empty group", () => {
+ const dag = new Dag("d");
+ const transform = dag.taskGroup("transform");
+ const cleaned = place(transform, "clean");
+ const nulls = place(transform.taskGroup("checks"), "nulls");
+ const publish = dag.taskGroup("publish");
+ place(publish, "push");
+ const extracted = place(dag, "extract");
+ const empty = dag.taskGroup("empty");
+ const loaded = place(dag, "load");
+ const reported = place(dag, "report");
+ cleaned.before(Label.of("rows").on(publish));
+ nulls.before(Label.of("checked").on(publish));
+ nulls.after(Label.of("pushed").on(publish));
+ extracted.before(Label.of("to empty").on(empty));
+ empty.before(loaded);
+ reported.after(Label.of("from empty").on(empty));
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ "transform.clean": { "publish.upstream_join_id": { label: "rows" } },
+ "transform.checks.nulls": { "publish.upstream_join_id": { label:
"checked" } },
+ "publish.downstream_join_id": { "transform.checks.nulls": { label:
"pushed" } },
+ extract: { "empty.upstream_join_id": { label: "to empty" } },
+ "empty.downstream_join_id": { report: { label: "from empty" } },
+ });
+ });
+
+ it("keeps labelling a root that a later edge inside the group puts behind
another task", () => {
+ const dag = new Dag("d");
+ const extracted = place(dag, "extract");
+ const g = dag.taskGroup("g");
+ const first = g.taskGroup("first");
+ place(first, "x");
+ const second = g.taskGroup("second");
+ place(second, "y");
+ extracted.before(Label.of("rows").on(g));
+ first.before(second);
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ extract: {
+ "g.upstream_join_id": { label: "rows" },
+ "g.first.x": { label: "rows" },
+ "g.second.y": { label: "rows" },
+ },
+ });
+ });
+
+ it("keeps labelling a leaf that a later edge inside the group puts before
another task", () => {
+ const dag = new Dag("d");
+ const g = dag.taskGroup("g");
+ const first = g.taskGroup("first");
+ place(first, "x");
+ const second = g.taskGroup("second");
+ place(second, "y");
+ const reported = place(dag, "report");
+ reported.after(Label.of("rows").on(g));
+ first.before(second);
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ "g.downstream_join_id": { report: { label: "rows" } },
+ "g.first.x": { report: { label: "rows" } },
+ "g.second.y": { report: { label: "rows" } },
+ });
+ });
+
+ it("labels no task that an earlier edge inside the group put behind
another task", () => {
+ const dag = new Dag("d");
+ const extracted = place(dag, "extract");
+ const outer = dag.taskGroup("outer");
+ const t = place(outer, "t");
+ const inner = outer.taskGroup("inner");
+ place(inner, "i");
+ inner.before(t);
+ extracted.before(Label.of("rows").on(outer));
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ extract: {
+ "outer.upstream_join_id": { label: "rows" },
+ "outer.inner.i": { label: "rows" },
+ },
+ });
+ });
+
+ it("gives a task edge the label set last, by the edge itself or by its
group edge", () => {
+ const dag = new Dag("d");
+ const extracted = place(dag, "extract");
+ const transform = dag.taskGroup("transform");
+ const cleaned = place(transform, "clean");
+ const validated = place(transform, "validate");
+ extracted.before(Label.of("rows").on(cleaned), validated);
+ extracted.before(Label.of("to transform").on(transform));
+ // Drawing this edge again without a label does not set a new label.
+ extracted.before(cleaned);
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ extract: {
+ "transform.clean": { label: "to transform" },
+ "transform.validate": { label: "to transform" },
+ "transform.upstream_join_id": { label: "to transform" },
+ },
+ });
+
+ extracted.before(Label.of("rows").on(cleaned));
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ extract: {
+ "transform.clean": { label: "rows" },
+ "transform.validate": { label: "to transform" },
+ "transform.upstream_join_id": { label: "to transform" },
+ },
+ });
+ });
+
+ it("lets a later group edge relabel a task edge that an earlier one
labelled", () => {
+ const dag = new Dag("d");
+ const extracted = place(dag, "extract");
+ const outer = dag.taskGroup("outer");
+ const inner = outer.taskGroup("inner");
+ place(inner, "i");
+ extracted.before(Label.of("a").on(outer));
+ extracted.before(Label.of("b").on(inner));
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ extract: {
+ "outer.inner.i": { label: "b" },
+ "outer.upstream_join_id": { label: "a" },
+ "outer.inner.upstream_join_id": { label: "b" },
+ },
+ });
+
+ extracted.before(Label.of("c").on(outer));
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ extract: {
+ "outer.inner.i": { label: "c" },
+ "outer.upstream_join_id": { label: "c" },
+ "outer.inner.upstream_join_id": { label: "b" },
+ },
+ });
+ });
+
it("counts one edge when the wiring already drew it", () => {
const dag = new Dag("d");
const extracted = place(dag, "extract");
diff --git a/ts-sdk/tests/public-api.test.ts b/ts-sdk/tests/public-api.test.ts
index ff19a660fc6..163cb14471a 100644
--- a/ts-sdk/tests/public-api.test.ts
+++ b/ts-sdk/tests/public-api.test.ts
@@ -49,7 +49,7 @@ import {
Dag,
getClient,
getContext,
- label,
+ Label,
NEVER_EXPIRE,
SUPERVISOR_API_VERSION,
TaskHandler,
@@ -73,7 +73,13 @@ describe("public API", () => {
taskId: "public_api_downstream",
});
expect(dag.taskIds).toEqual(["public_api_task", "public_api_downstream"]);
- expect(() => upstream.before(label(downstream, "rows"))).not.toThrow();
+ expect(() =>
upstream.before(Label.of("rows").on(downstream))).not.toThrow();
+ expectTypeOf<keyof Label>().toEqualTypeOf<"on">();
+ const rejectsLabelWithoutNode = () => {
+ // @ts-expect-error a label needs on() to name the node that its edge
connects to.
+ upstream.before(Label.of("rows"));
+ };
+ void rejectsLabelWithoutNode;
// serve() hands the bundle to the runtime, which needs the supervisor's
// socket addresses that Airflow puts on argv.
await expect(new Bundle(dag).serve()).rejects.toThrow("Missing --comm");
diff --git a/ts-sdk/tests/sdk/branching.test.ts
b/ts-sdk/tests/sdk/branching.test.ts
index 1c060ea5c18..9245b38b879 100644
--- a/ts-sdk/tests/sdk/branching.test.ts
+++ b/ts-sdk/tests/sdk/branching.test.ts
@@ -24,7 +24,7 @@ import {
getDagOrderEdges,
getDagTaskInputs,
getDagTaskRecords,
- label,
+ Label,
type TaskRef,
} from "../../src/sdk/dag.js";
import { serializeDag } from "../../src/coordinator/serde.js";
@@ -197,9 +197,9 @@ describe("dag.switch", () => {
it("a labelled case", () => {
const { long, decider } = branchedDag();
- expect(() => decider.case(label(long, "long") as never)).toThrowError(
- 'A case of "pick_path" of Dag "d" was given a label(); label a branch
edge ' +
- 'by redrawing it from "pick_path" with before(label(task, "text"))',
+ expect(() => decider.case(Label.of("long").on(long) as
never)).toThrowError(
+ 'A case of "pick_path" of Dag "d" was given Label.of(...).on(...);
label a branch ' +
+ 'edge by redrawing it from "pick_path" with
before(Label.of("text").on(task))',
);
});
diff --git a/ts-sdk/tests/sdk/conditional.test.ts
b/ts-sdk/tests/sdk/conditional.test.ts
index 7cb5e7888aa..71f8656ddc7 100644
--- a/ts-sdk/tests/sdk/conditional.test.ts
+++ b/ts-sdk/tests/sdk/conditional.test.ts
@@ -24,7 +24,7 @@ import {
getDagOrderEdges,
getDagTaskInputs,
getDagTaskRecords,
- label,
+ Label,
type TaskRef,
} from "../../src/sdk/dag.js";
import { serializeDag } from "../../src/coordinator/serde.js";
@@ -236,7 +236,7 @@ describe("dag.if", () => {
const { dag, ifReady, gated } = gatedDag();
gated.then(ifReady);
- gated.before(label(ifReady, "rows found"));
+ gated.before(Label.of("rows found").on(ifReady));
expect(getDagOrderEdges(dag)).toContainEqual({
upstream: "has_rows",
@@ -249,9 +249,9 @@ describe("dag.if", () => {
it("a labelled branch", () => {
const { ifReady, gated } = gatedDag();
- expect(() => gated.then(label(ifReady, "rows found") as
never)).toThrowError(
- 'The "then" branch of "has_rows" of Dag "d" was given a label(); label
a branch edge ' +
- 'by redrawing it from "has_rows" with before(label(task, "text"))',
+ expect(() => gated.then(Label.of("rows found").on(ifReady) as
never)).toThrowError(
+ 'The "then" branch of "has_rows" of Dag "d" was given
Label.of(...).on(...); label a ' +
+ 'branch edge by redrawing it from "has_rows" with
before(Label.of("text").on(task))',
);
});
diff --git a/ts-sdk/tests/sdk/dag.test.ts b/ts-sdk/tests/sdk/dag.test.ts
index c0f1be41e3f..96cbe4b3ae9 100644
--- a/ts-sdk/tests/sdk/dag.test.ts
+++ b/ts-sdk/tests/sdk/dag.test.ts
@@ -25,7 +25,7 @@ import {
getDagTaskGroups,
getDagTaskInputs,
getDagTaskRecords,
- label,
+ Label,
type DagSpec,
type TaskRef,
type TaskSpec,
@@ -566,8 +566,11 @@ describe("Dag", () => {
it("labels each edge of a fan-out from its own endpoint", () => {
const { dag, refs } = placedDag("d", "check", "process", "notify");
- refs.check!.before(label(refs.process!, "rows found"),
label(refs.notify!, "no rows"));
- refs.notify!.after(label(refs.process!, "after processing"));
+ refs.check!.before(
+ Label.of("rows found").on(refs.process!),
+ Label.of("no rows").on(refs.notify!),
+ );
+ refs.notify!.after(Label.of("after processing").on(refs.process!));
expect(getDagOrderEdges(dag)).toEqual([
{ upstream: "check", downstream: "process", label: "rows found" },
@@ -576,11 +579,44 @@ describe("Dag", () => {
]);
});
+ it("puts one label on the edge to or from each node it is on", () => {
+ const { dag, refs } = placedDag("d", "load", "report_empty", "notify",
"audit");
+ const staging = dag.taskGroup("staging");
+ staging.task("stage", async () => undefined)();
+
+ refs.load!.before(Label.of("forks").on(refs.report_empty!, staging));
+ refs.audit!.after(Label.of("done").on(refs.report_empty!, refs.notify!));
+
+ expect(getDagOrderEdges(dag)).toEqual([
+ { upstream: "load", downstream: "report_empty", label: "forks" },
+ { upstream: "load", downstream: "staging", label: "forks" },
+ { upstream: "report_empty", downstream: "audit", label: "done" },
+ { upstream: "notify", downstream: "audit", label: "done" },
+ ]);
+ });
+
+ it("gives a labelled node that is labelled again the outer label", () => {
+ const { dag, refs } = placedDag("d", "left", "middle", "right", "join");
+
+ refs.join!.after(
+ Label.of("done").on(
+ Label.of("inner").on(refs.left!),
+ Label.of("inner").on(refs.middle!, refs.right!),
+ ),
+ );
+
+ expect(getDagOrderEdges(dag)).toEqual([
+ { upstream: "left", downstream: "join", label: "done" },
+ { upstream: "middle", downstream: "join", label: "done" },
+ { upstream: "right", downstream: "join", label: "done" },
+ ]);
+ });
+
it("replaces a label on a redrawn edge, and keeps it on an unlabelled
redraw", () => {
const { dag, refs } = placedDag("d", "load", "cleanup");
- refs.load!.before(label(refs.cleanup!, "first"));
- refs.load!.before(label(refs.cleanup!, "second"));
+ refs.load!.before(Label.of("first").on(refs.cleanup!));
+ refs.load!.before(Label.of("second").on(refs.cleanup!));
refs.cleanup!.after(refs.load!);
expect(getDagOrderEdges(dag)).toEqual([
@@ -597,28 +633,68 @@ describe("Dag", () => {
});
it.each(["before", "after"] as const)("rejects %s() called on a label",
(verb) => {
- const { refs } = placedDag("d", "load", "cleanup");
+ const { refs } = placedDag("d", "load", "audit", "cleanup");
- expect(() => label(refs.load!,
"rows")[verb](refs.cleanup!)).toThrowError(
- `label(..., "rows") on "load" of Dag "d" names the edge drawn to it, `
+
+ expect(() =>
Label.of("rows").on(refs.load!)[verb](refs.cleanup!)).toThrowError(
+ `Label.of("rows") on "load" of Dag "d" labels edges, ` +
+ `so pass it to before() or after() rather than calling ${verb}() on
it`,
+ );
+ expect(() => Label.of("rows").on(refs.load!,
refs.audit!)[verb](refs.cleanup!)).toThrowError(
+ `Label.of("rows") on "load", "audit" of Dag "d" labels edges, ` +
`so pass it to before() or after() rather than calling ${verb}() on
it`,
);
});
+ it("rejects a label on nodes of two Dags", () => {
+ const { refs } = placedDag("d", "load");
+ const other = placedDag("e", "cleanup").refs;
+
+ expect(() => Label.of("rows").on(refs.load!,
other.cleanup!)).toThrowError(
+ 'Label.of("rows").on() takes nodes of one Dag, but "load" is in Dag
"d" and "cleanup" ' +
+ 'is in Dag "e"',
+ );
+ });
+
+ it("draws none of a label's edges when one of them cannot be drawn", () =>
{
+ const { dag, refs } = placedDag("d", "load", "cleanup");
+
+ expect(() => refs.load!.before(Label.of("rows").on(refs.cleanup!,
refs.load!))).toThrowError(
+ /cannot draw an edge from node "load" of Dag "d" to itself/,
+ );
+ expect(getDagOrderEdges(dag)).toEqual([]);
+ });
+
it.each([
["undefined", undefined],
["an object", { dagId: "d", taskId: "load" }],
])("rejects a label on %s", (described, node) => {
- expect(() => label(node as never, "rows")).toThrowError(
- `label() takes a task or task group reference, not ${described}`,
+ const { refs } = placedDag("d", "load");
+ const expected = `Label.of("rows").on() takes a task or task group
reference, not ${described}`;
+
+ expect(() => Label.of("rows").on(node as never)).toThrowError(expected);
+ expect(() => Label.of("rows").on(refs.load!, node as
never)).toThrowError(expected);
+ });
+
+ it("rejects a label on no node", () => {
+ // @ts-expect-error on() needs a node, but plain JavaScript can call
on() with no node.
+ expect(() => Label.of("rows").on()).toThrowError(
+ 'Label.of("rows").on() takes a task or task group reference, not
undefined',
);
});
- it("rejects an empty label", () => {
- const { refs } = placedDag("d", "load");
+ it.each([
+ ["an empty string", ""],
+ ["a blank string", " \t\n"],
+ ])("rejects %s as a label", (described, text) => {
+ expect(() => Label.of(text)).toThrowError(
+ `Label.of() takes the text to put on the edge, not ${described}`,
+ );
+ });
- expect(() => label(refs.load!, "")).toThrowError(
- "label() takes the text to put on the edge, not an empty string",
+ it("rejects an empty label made with new, which plain JavaScript can
call", () => {
+ // @ts-expect-error the constructor is private, so TypeScript code must
use Label.of.
+ expect(() => new Label("")).toThrowError(
+ "Label.of() takes the text to put on the edge, not an empty string",
);
});
@@ -627,18 +703,25 @@ describe("Dag", () => {
const extracted = dag.task("extract", async () => undefined)();
const load = dag.task("load", async (_: { rows: unknown }) => undefined);
- expect(() => load({ rows: label(extracted, "rows") as never
})).toThrowError(
- 'Input "rows" of task "load" was given a label(), which only goes into
before() or ' +
- "after(); pass the reference itself",
+ expect(() => load({ rows: Label.of("rows").on(extracted) as never
})).toThrowError(
+ 'Input "rows" of task "load" was given Label.of(...).on(...), which
only goes into ' +
+ "before() or after(); pass the reference itself",
);
});
- it("rejects a label nested in a literal task input", () => {
+ it.each([
+ ["the task", false],
+ ["a group and then the task", true],
+ ])("rejects a label on %s nested in a literal task input", (_label,
withGroup) => {
const dag = new Dag("d");
const extracted = dag.task("extract", async () => undefined)();
+ const staging = dag.taskGroup("staging");
const load = dag.task("load", async (_: { rows: unknown }) => undefined);
+ const labelled = withGroup
+ ? Label.of("rows").on(staging, extracted)
+ : Label.of("rows").on(extracted);
- expect(() => load({ rows: [label(extracted, "rows")] as never
})).toThrowError(
+ expect(() => load({ rows: [labelled] as never })).toThrowError(
/Input "rows" of task "load" holds a reference to "extract" inside a
literal value/,
);
});
@@ -650,17 +733,15 @@ describe("Dag", () => {
expect(() =>
dag.if(
async (_: { rows: unknown }) => true,
- { rows: label(extracted, "rows") as never },
+ { rows: Label.of("rows").on(extracted) as never },
{ taskId: "has_rows" },
),
- ).toThrowError('Input "rows" of task "has_rows" was given a label()');
+ ).toThrowError('Input "rows" of task "has_rows" was given
Label.of(...).on(...)');
});
it("rejects a label that is not a string", () => {
- const { refs } = placedDag("d", "load");
-
- expect(() => label(refs.load!, 42 as never)).toThrowError(
- "label() takes a string to name the edge with, not the number 42",
+ expect(() => Label.of(42 as never)).toThrowError(
+ "Label.of() takes a string to name the edge with, not the number 42",
);
});
@@ -670,7 +751,7 @@ describe("Dag", () => {
Object.defineProperty(foreign,
Symbol.for("airflow.ts-sdk.LabelledNode"), { value: true });
expect(() => refs.load!.before(foreign)).toThrowError(
- /before\(\) on Dag "d" was given a label\(\) that comes from a
different copy/,
+ /before\(\) on Dag "d" was given Label\.of\(\.\.\.\)\.on\(\.\.\.\),
which comes from a different copy/,
);
});
