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 75b8c05bb88 [v3-4-test] TS SDK: Add edge labels with label() (#74494)
(#75031)
75b8c05bb88 is described below
commit 75b8c05bb8862a399944fbace7205096096b752a
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 19 15:15:17 2026 +0530
[v3-4-test] TS SDK: Add edge labels with label() (#74494) (#75031)
* TS SDK: Add edge labels with label()
* TS SDK: Tighten label() misuse checks
- Throw when before() or after() is called on a label, instead of dropping
it.
- Validate label() arguments and reject labels in then/else/case with a
hint.
- Report a label from another SDK copy with the duplicate-copy hint.
- Omit the label key on unlabelled edges.
- Test a labelled group-as-upstream edge on an acyclic Dag.
* Use US spelling of labeled in TS SDK docs
* TS SDK: Reject misplaced and empty labels, reserve group join IDs
A label() passed as a task input drew no edge and failed later at
serialization, and an empty label replaced an earlier one on the edge.
A task or group could also take a group's join node ID, under which
edge labels at a group end are keyed. label is now in the API docs.
---------
(cherry picked from commit 777bf9b18881f59a9fb2bfd55f033dbe8904d15c)
Co-authored-by: Guan-Ming Chiu
<[email protected]>
Co-authored-by: ZHE YOU LIU <[email protected]>
Co-authored-by: Claude <[email protected]>
---
.../language-sdks/typescript.rst | 18 +++
ts-sdk/api-docs/dag-authoring-api.ts | 1 +
ts-sdk/src/coordinator/serde.ts | 18 ++-
ts-sdk/src/index.ts | 2 +-
ts-sdk/src/sdk/dag.ts | 112 ++++++++++++++++--
ts-sdk/tests/coordinator/serde.test.ts | 19 +++
ts-sdk/tests/public-api.test.ts | 2 +
ts-sdk/tests/sdk/branching.test.ts | 10 ++
ts-sdk/tests/sdk/conditional.test.ts | 23 ++++
ts-sdk/tests/sdk/dag.test.ts | 128 +++++++++++++++++++++
10 files changed, 321 insertions(+), 12 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 0cb37f578e2..84b40c60ca9 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -243,6 +243,24 @@ 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:
+
+.. code-block:: typescript
+
+ import { label } from "apache-airflow-ts-sdk";
+
+ checked.before(label(processed, "rows found"), label(notified, "no rows"));
+
+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):
+
+.. code-block:: typescript
+
+ const rowCheck = dag.if(hasRows, { total });
+ rowCheck.then(loaded).else(reportedEmpty);
+ rowCheck.before(label(loaded, "rows found"), label(reportedEmpty, "no
rows"));
+
Task groups
~~~~~~~~~~~
diff --git a/ts-sdk/api-docs/dag-authoring-api.ts
b/ts-sdk/api-docs/dag-authoring-api.ts
index b37278f7f99..8fd6cc4427c 100644
--- a/ts-sdk/api-docs/dag-authoring-api.ts
+++ b/ts-sdk/api-docs/dag-authoring-api.ts
@@ -24,6 +24,7 @@ export {
Dag,
getClient,
getContext,
+ label,
NEVER_EXPIRE,
TaskHandler,
triggerDagRun,
diff --git a/ts-sdk/src/coordinator/serde.ts b/ts-sdk/src/coordinator/serde.ts
index 95da4e364af..c8bcd655009 100644
--- a/ts-sdk/src/coordinator/serde.ts
+++ b/ts-sdk/src/coordinator/serde.ts
@@ -41,10 +41,13 @@ import {
} from "../generated/dag-schema-fields.js";
import type { JsonValue } from "../sdk/client-types.js";
import {
+ DOWNSTREAM_JOIN_SUFFIX,
+ getDagOrderEdges,
getDagTaskGroups,
getDagTaskInputs,
getDagTaskRecords,
isTaskRef,
+ UPSTREAM_JOIN_SUFFIX,
type Dag,
type RecordedInputs,
type TaskGroupRecord,
@@ -84,6 +87,19 @@ 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 {
+ const groups = getDagTaskGroups(dag);
+ const info: Record<string, Record<string, SerializedValue>> = {};
+ for (const { upstream, downstream, label } of getDagOrderEdges(dag)) {
+ if (label === undefined) continue;
+ const from = groups.has(upstream) ? upstream + DOWNSTREAM_JOIN_SUFFIX :
upstream;
+ const to = groups.has(downstream) ? downstream + UPSTREAM_JOIN_SUFFIX :
downstream;
+ (info[from] ??= {})[to] = { label };
+ }
+ return info;
+}
+
/** The Dags this one's tasks depend on, for the UI dependency graph. */
function serializeDagDependencies(dag: Dag): SerializedValue {
const dependencies: SerializedValue[] = [];
@@ -170,7 +186,7 @@ export function serializeDag(
),
dag_dependencies: serializeDagDependencies(dag),
task_group: serializeTaskGroups(dag, graph),
- edge_info: {},
+ edge_info: serializeEdgeInfo(dag),
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 4c629bda76a..865eae75df2 100644
--- a/ts-sdk/src/index.ts
+++ b/ts-sdk/src/index.ts
@@ -17,7 +17,7 @@
* under the License.
*/
-export { Dag } 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.ts b/ts-sdk/src/sdk/dag.ts
index d31b5f2fe8d..398d5fc7fb3 100644
--- a/ts-sdk/src/sdk/dag.ts
+++ b/ts-sdk/src/sdk/dag.ts
@@ -270,6 +270,7 @@ export interface TaskGroupOptions {
export interface OrderEdge {
readonly upstream: string;
readonly downstream: string;
+ readonly label?: string;
}
// A task id cannot hold a NUL, so a joined pair cannot collide with one.
@@ -292,6 +293,15 @@ export interface TaskGroupRecord {
/** Separates a group ID from what it contains, as Python's `prefix_group_id`
does. */
const GROUP_SEPARATOR = ".";
+/**
+ * Internal: suffixes of the join nodes the Airflow UI draws at each end of a
+ * task group, as Python's `TaskGroup.upstream_join_id` and
+ * `downstream_join_id` name them.
+ */
+export const UPSTREAM_JOIN_SUFFIX = ".upstream_join_id";
+export const DOWNSTREAM_JOIN_SUFFIX = ".downstream_join_id";
+const JOIN_SUFFIXES = [UPSTREAM_JOIN_SUFFIX, DOWNSTREAM_JOIN_SUFFIX];
+
/** The Dag's own view of a group, which it appends to as an author declares.
*/
interface MutableTaskGroupRecord extends TaskGroupRecord {
readonly taskIds: string[];
@@ -317,8 +327,52 @@ 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 }>();
+
function resolveNode(node: Node): Node {
- return (typeof node === "object" && node !== null &&
conditionTasks.get(node)) || node;
+ if (typeof node !== "object" || node === null) return node;
+ return labelledNodes.get(node)?.node ?? 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.
+ *
+ * ```ts
+ * checked.before(label(processed, "rows found"), label(emptyNotice, "no
rows"));
+ * ```
+ */
+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)}`);
+ }
+ if (text === "") {
+ throw new Error("label() takes the text to put on the edge, not an empty
string");
+ }
+ 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)}`);
+ }
+ 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 {
+ return hasBrand(value, "LabelledNode");
}
/**
@@ -334,6 +388,8 @@ function resolveNode(node: Node): Node {
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;
// 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;
@@ -693,6 +749,7 @@ export class Dag {
"a condition names each side once",
);
}
+ this.#rejectLabel(target, `The "${side}" branch of "${taskId}"`, taskId);
const taskRef = resolveNode(target) as TaskRef;
this.#validateOwnNode(taskRef, `the "${side}" branch of "${taskId}"`);
if (!isTaskRef(taskRef)) {
@@ -800,6 +857,7 @@ export class Dag {
dagId: this.dagId,
taskId,
case: (taskRef) => {
+ this.#rejectLabel(taskRef, `A case of "${taskId}"`, taskId);
this.#validateOwnNode(taskRef, `a case of "${taskId}"`);
if (!isTaskRef(taskRef)) {
throw new Error(
@@ -1070,11 +1128,21 @@ export class Dag {
// Tasks and groups share one namespace, as they do in Python: a serialized
// Dag addresses both by a bare ID, so `dag.task("x")` and
`dag.taskGroup("x")`
- // cannot both exist.
+ // cannot both exist. A group's join node IDs are in it too, since edge
labels
+ // at a group end are keyed by them.
#reserveNodeId(id: string, kind: "Task" | "Task group"): void {
if (this.#tasks.has(id) || this.#groups.has(id)) {
throw new Error(`${kind} "${id}" is already registered for Dag
"${this.dagId}"`);
}
+ for (const suffix of JOIN_SUFFIXES) {
+ const groupId = id.endsWith(suffix) ? id.slice(0, -suffix.length) :
undefined;
+ if (groupId !== undefined && this.#groups.has(groupId)) {
+ throw new Error(
+ `${kind} "${id}" of Dag "${this.dagId}" takes the ID of a join node
of task group ` +
+ `"${groupId}"; pick another ID`,
+ );
+ }
+ }
}
#createTaskGroupRef(groupId: string): TaskGroupRef {
@@ -1121,6 +1189,7 @@ export class Dag {
}
#addOrderEdge(upstreamNode: Node, downstreamNode: Node, verb: "before" |
"after"): void {
+ const text = labelledNodes.get(verb === "before" ? downstreamNode :
upstreamNode)?.text;
const upstream = resolveNode(upstreamNode);
const downstream = resolveNode(downstreamNode);
if (this.#finalized) {
@@ -1143,25 +1212,42 @@ export class Dag {
}
const key = `${upstreamId}${EDGE_KEY_SEPARATOR}${downstreamId}`;
// Idempotent, so an edge drawn from both ends is one edge.
- if (!this.#orderEdges.has(key)) {
- this.#orderEdges.set(
- key,
- Object.freeze({ upstream: upstreamId!, downstream: downstreamId! }),
+ const edgeLabel = text ?? this.#orderEdges.get(key)?.label;
+ this.#orderEdges.set(
+ key,
+ Object.freeze({
+ upstream: upstreamId!,
+ downstream: downstreamId!,
+ ...(edgeLabel === undefined ? {} : { label: edgeLabel }),
+ }),
+ );
+ }
+
+ #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"))`,
);
}
}
- #validateOwnNode(node: Node, label: string): void {
+ #validateOwnNode(node: Node, context: string): void {
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}`,
+ );
+ }
if (id === undefined) {
throw new Error(
- `${label} on Dag "${this.dagId}" takes tasks and task groups this Dag
handed out, ` +
+ `${context} on Dag "${this.dagId}" takes tasks and task groups this
Dag handed out, ` +
"not arbitrary values",
);
}
if (node.dagId !== this.dagId) {
throw new Error(
- `${label} cannot reach Dag "${node.dagId}" node "${id}" from Dag ` +
+ `${context} cannot reach Dag "${node.dagId}" node "${id}" from Dag ` +
`"${this.dagId}"; an edge joins two nodes of one Dag`,
);
}
@@ -1171,7 +1257,7 @@ export class Dag {
// the edge at this Dag's own group of that name.
if (isTaskRef(node) ? this.#tasks.get(id)?.task !== node :
this.#groupRefs.get(id) !== node) {
throw new Error(
- `${label} was given a reference to "${id}" that this Dag did not hand
out; ` +
+ `${context} was given a reference to "${id}" that this Dag did not
hand out; ` +
`it comes from another Dag object with the same ID, or
${DUPLICATE_COPY_HINT}`,
);
}
@@ -1217,6 +1303,12 @@ 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",
+ );
+ }
// Anything unbranded is a literal argument, including a look-alike
// `{dagId, taskId}` object: only a real reference makes an edge.
if (!isTaskRef(value)) {
diff --git a/ts-sdk/tests/coordinator/serde.test.ts
b/ts-sdk/tests/coordinator/serde.test.ts
index aa7f533c855..0b15d715750 100644
--- a/ts-sdk/tests/coordinator/serde.test.ts
+++ b/ts-sdk/tests/coordinator/serde.test.ts
@@ -27,6 +27,7 @@ import {
} from "../../src/coordinator/serde.js";
import {
Dag,
+ label,
type DagSpec,
type TaskGroupRef,
type TaskRef,
@@ -485,6 +486,24 @@ describe("serializeDag", () => {
]);
});
+ it("writes edge labels keyed by the group join nodes at a group end", ()
=> {
+ const dag = new Dag("d");
+ const extracted = place(dag, "extract");
+ const transformed = place(dag, "transform", [extracted]);
+ 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"));
+
+ expect((serializeDag(dag, "", ".") as Json)["edge_info"]).toEqual({
+ extract: { transform: { label: "rows" } },
+ transform: { "staging.upstream_join_id": { label: "staged" } },
+ "staging.downstream_join_id": { load: { label: "ready" } },
+ });
+ });
+
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 1fc2d9bc4fd..ff19a660fc6 100644
--- a/ts-sdk/tests/public-api.test.ts
+++ b/ts-sdk/tests/public-api.test.ts
@@ -49,6 +49,7 @@ import {
Dag,
getClient,
getContext,
+ label,
NEVER_EXPIRE,
SUPERVISOR_API_VERSION,
TaskHandler,
@@ -72,6 +73,7 @@ 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();
// 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 fb31c2022ee..1c060ea5c18 100644
--- a/ts-sdk/tests/sdk/branching.test.ts
+++ b/ts-sdk/tests/sdk/branching.test.ts
@@ -24,6 +24,7 @@ import {
getDagOrderEdges,
getDagTaskInputs,
getDagTaskRecords,
+ label,
type TaskRef,
} from "../../src/sdk/dag.js";
import { serializeDag } from "../../src/coordinator/serde.js";
@@ -193,6 +194,15 @@ describe("dag.switch", () => {
});
describe("rejects", () => {
+ 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"))',
+ );
+ });
+
it("a task group as a case", () => {
const { dag, decider } = branchedDag();
const group = dag.taskGroup("staging");
diff --git a/ts-sdk/tests/sdk/conditional.test.ts
b/ts-sdk/tests/sdk/conditional.test.ts
index f2487f2af6a..7cb5e7888aa 100644
--- a/ts-sdk/tests/sdk/conditional.test.ts
+++ b/ts-sdk/tests/sdk/conditional.test.ts
@@ -24,6 +24,7 @@ import {
getDagOrderEdges,
getDagTaskInputs,
getDagTaskRecords,
+ label,
type TaskRef,
} from "../../src/sdk/dag.js";
import { serializeDag } from "../../src/coordinator/serde.js";
@@ -231,7 +232,29 @@ describe("dag.if", () => {
});
});
+ it("labels a branch edge redrawn from the condition", () => {
+ const { dag, ifReady, gated } = gatedDag();
+ gated.then(ifReady);
+
+ gated.before(label(ifReady, "rows found"));
+
+ expect(getDagOrderEdges(dag)).toContainEqual({
+ upstream: "has_rows",
+ downstream: "load_if_ready",
+ label: "rows found",
+ });
+ });
+
describe("rejects", () => {
+ 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"))',
+ );
+ });
+
it("a task group as a branch", () => {
const { dag, gated } = gatedDag();
const group = dag.taskGroup("staging");
diff --git a/ts-sdk/tests/sdk/dag.test.ts b/ts-sdk/tests/sdk/dag.test.ts
index ab9e4b902fb..c0f1be41e3f 100644
--- a/ts-sdk/tests/sdk/dag.test.ts
+++ b/ts-sdk/tests/sdk/dag.test.ts
@@ -25,6 +25,7 @@ import {
getDagTaskGroups,
getDagTaskInputs,
getDagTaskRecords,
+ label,
type DagSpec,
type TaskRef,
type TaskSpec,
@@ -562,6 +563,117 @@ 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"));
+
+ expect(getDagOrderEdges(dag)).toEqual([
+ { upstream: "check", downstream: "process", label: "rows found" },
+ { upstream: "check", downstream: "notify", label: "no rows" },
+ { upstream: "process", downstream: "notify", label: "after processing"
},
+ ]);
+ });
+
+ 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.cleanup!.after(refs.load!);
+
+ expect(getDagOrderEdges(dag)).toEqual([
+ { upstream: "load", downstream: "cleanup", label: "second" },
+ ]);
+ });
+
+ it("leaves the label out of an unlabelled edge", () => {
+ const { dag, refs } = placedDag("d", "load", "cleanup");
+
+ refs.load!.before(refs.cleanup!);
+
+ expect(getDagOrderEdges(dag)).toStrictEqual([{ upstream: "load",
downstream: "cleanup" }]);
+ });
+
+ it.each(["before", "after"] as const)("rejects %s() called on a label",
(verb) => {
+ const { refs } = placedDag("d", "load", "cleanup");
+
+ expect(() => label(refs.load!,
"rows")[verb](refs.cleanup!)).toThrowError(
+ `label(..., "rows") on "load" of Dag "d" names the edge drawn to it, `
+
+ `so pass it to before() or after() rather than calling ${verb}() on
it`,
+ );
+ });
+
+ 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}`,
+ );
+ });
+
+ it("rejects an empty label", () => {
+ const { refs } = placedDag("d", "load");
+
+ expect(() => label(refs.load!, "")).toThrowError(
+ "label() takes the text to put on the edge, not an empty string",
+ );
+ });
+
+ it("rejects a label given as a task input", () => {
+ const dag = new Dag("d");
+ 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",
+ );
+ });
+
+ it("rejects a label nested in a literal task input", () => {
+ const dag = new Dag("d");
+ 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" holds a reference to "extract" inside a
literal value/,
+ );
+ });
+
+ it("rejects a label given as a decider input", () => {
+ const dag = new Dag("d");
+ const extracted = dag.task("extract", async () => undefined)();
+
+ expect(() =>
+ dag.if(
+ async (_: { rows: unknown }) => true,
+ { rows: label(extracted, "rows") as never },
+ { taskId: "has_rows" },
+ ),
+ ).toThrowError('Input "rows" of task "has_rows" was given a label()');
+ });
+
+ 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",
+ );
+ });
+
+ it("rejects a label from another copy of the SDK", () => {
+ const { refs } = placedDag("d", "load", "cleanup");
+ const foreign = { dagId: "d", before: () => foreign, after: () =>
foreign };
+ 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/,
+ );
+ });
+
it.each([
["before", (a: TaskRef, b: TaskRef) => a.before(b)],
["after", (a: TaskRef, b: TaskRef) => b.after(a)],
@@ -864,6 +976,22 @@ describe("Dag", () => {
},
);
+ it.each(["upstream_join_id", "downstream_join_id"])(
+ "rejects a task or group taking the %s of its group",
+ (joinId) => {
+ const dag = new Dag("grouped");
+ const staging = dag.taskGroup("staging");
+
+ expect(() => staging.task(joinId, async () => undefined)).toThrowError(
+ `Task "staging.${joinId}" of Dag "grouped" takes the ID of a join
node of task group ` +
+ '"staging"; pick another ID',
+ );
+ expect(() => staging.taskGroup(joinId)).toThrowError(
+ `Task group "staging.${joinId}" of Dag "grouped" takes the ID of a
join node`,
+ );
+ },
+ );
+
it("allows the same group ID under different parents", () => {
const dag = new Dag("grouped");
dag
