Skip to content

Commit f9be893

Browse files
committed
TS SDK: Take a handler in dag.if
1 parent 4b00f64 commit f9be893

9 files changed

Lines changed: 200 additions & 234 deletions

File tree

‎airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -362,18 +362,22 @@ there. Set it once on the Dag and each task inherits it:
362362
Conditional branching
363363
~~~~~~~~~~~~~~~~~~~~~
364364

365-
``dag.if`` takes a task whose handler returns a boolean, and names the task each outcome runs:
365+
``dag.if`` takes a handler that returns a boolean, and names the task each outcome runs:
366366

367367
.. code-block:: typescript
368368
369-
const condition = dag.task("has_rows", async ({ rows }: { rows: number }) => rows > 0);
370-
const gated = condition({ rows: extracted });
369+
async function hasRows({ rows }: { rows: number }): Promise<boolean> {
370+
return rows > 0;
371+
}
371372
372-
dag.if(gated).then(loaded).else(reportedEmpty);
373+
const gate = dag.if(hasRows)({ rows: extracted });
374+
gate.then(loaded).else(reportedEmpty);
373375
374-
The condition is an ordinary task, so it is declared, typed and wired like any other, and the
375-
compiler checks that its handler really returns a boolean. ``else`` is optional: a one-sided
376-
condition skips its own branch when the condition fails and follows nothing.
376+
``dag.if`` declares the condition the way ``dag.task`` declares a task: its id is the function's
377+
name unless ``taskId`` sets it, and the returned factory takes its inputs. The compiler checks that
378+
the handler returns a boolean. ``else`` is optional: a one-sided condition skips its own branch
379+
when the condition fails and follows nothing. The condition is also a node, so ``notified.after(gate)``
380+
orders a task after it.
377381

378382
A guarded task takes no argument for the control edge, because a condition's boolean decides whether
379383
the task runs rather than what it runs on. Read a value from the condition with

‎ts-sdk/adr/0002-native-dag-interface.md‎

Lines changed: 15 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -142,18 +142,24 @@ convention.
142142
names after the host language's control flow:
143143

144144
```ts
145-
const gated = dag.task("has_rows", async ({ rows }: { rows: number }) => rows > 0)({
146-
rows: extracted,
147-
});
145+
async function hasRows({ rows }: { rows: number }): Promise<boolean> {
146+
return rows > 0;
147+
}
148148

149-
dag.if(gated).then(loadIfReady).else(loadFallback);
149+
const gate = dag.if(hasRows)({ rows: validated });
150+
gate.then(loadIfReady).else(loadFallback);
150151
```
151152

152-
**The condition is a task reference, not a task id and a function.** `dag.if` takes a `TaskRef` the
153-
Dag already handed back, whose handler's return type the compiler checks is `boolean`. So nothing
154-
depends on the task's id or on its function name, and a condition is declared, named and typed the
155-
same way every other task is. The same reasoning that makes a *case* a reference in that ADR's
156-
decision 2 applies to the condition itself.
153+
**The condition is a handler, declared the way a task is.** That ADR writes
154+
`dag.If(hasRows, airflow.Inputs(validated))` in Go, where `dag.Task(fn, airflow.Inputs(x))` declares
155+
and wires a task in one call. TypeScript splits that call into `dag.task(fn)({ x })`, so `dag.if`
156+
splits the same way: `dag.if(hasRows)` declares the deciding task, taking its id from the function's
157+
name unless `taskId` sets it, and the returned factory takes its inputs. The compiler checks that the
158+
handler returns a `boolean`.
159+
160+
**A condition is a node.** What the factory returns carries `before` and `after`, and stands at the
161+
other end of an edge as the deciding task, so `notified.after(gate)` reads as Go's `.After(gate)`. It
162+
is also a valid branch of another condition.
157163

158164
**A `then` chain is a thenable, and is guarded rather than avoided.** An object with a callable
159165
`then` is a *thenable*: were the object `dag.if` returns to reach an `await`, the runtime would hand

‎ts-sdk/src/coordinator/client.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,9 @@ export interface CoordinatorClient extends TaskClient {
8383
* output fails the task, while one that pushed null binds null.
8484
*/
8585
getXComEntry(opts: GetXComOpts): Promise<XComEntry>;
86+
87+
/** Mark direct downstream tasks of the running task as skipped; none is a no-op. */
88+
skipDownstreamTasks(taskIds: readonly string[]): Promise<void>;
8689
}
8790

8891
export function createCoordinatorClient(
@@ -206,8 +209,6 @@ export function createCoordinatorClient(
206209
// ---- Control flow ----
207210

208211
async skipDownstreamTasks(taskIds: readonly string[]): Promise<void> {
209-
// The supervisor treats an empty list as "skip nothing", and sending it
210-
// would cost a round trip to say so.
211212
if (taskIds.length === 0) return;
212213
const msg: SkipDownstreamTasks = { type: "SkipDownstreamTasks", tasks: [...taskIds] };
213214
await rpc("SkipDownstreamTasks", null, msg, () => undefined, "throw");

‎ts-sdk/src/coordinator/serde.ts‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -200,9 +200,7 @@ function serializeTask(
200200
const label = `task "${taskId}" of Dag "${dagId}"`;
201201
const bindings = serializeArgBindings(inputs, label);
202202
if (bindings) data["_arg_bindings"] = bindings;
203-
// What Python writes for a SkipMixin operator, and what makes
204-
// `NotPreviouslySkippedDep` consult this task's `skipmixin_key` XCom when one
205-
// of its skipped downstream tasks is cleared.
203+
// Lets `NotPreviouslySkippedDep` re-skip a cleared downstream, as Python's SkipMixin does.
206204
if (canSkipDownstream) data["_can_skip_downstream"] = true;
207205
applySchemaFields(data, spec, TASK_FIELD_RULES, label);
208206
if (downstream?.size) {

‎ts-sdk/src/index.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ export type { Registerable } from "./sdk/bundle.js";
2929
export type {
3030
Condition,
3131
ConditionElse,
32+
ConditionFactory,
3233
DagSpec,
3334
Node,
3435
TaskFactory,

‎ts-sdk/src/sdk/client.ts‎

Lines changed: 0 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -107,19 +107,6 @@ export interface TaskClient {
107107
* @throws {@link Exceptions!ConnectionNotFoundError | ConnectionNotFoundError} when the connection does not exist.
108108
*/
109109
getConnectionOrThrow(connId: string): Promise<ConnectionResult>;
110-
111-
/**
112-
* Mark downstream tasks of the running task as skipped.
113-
*
114-
* The mechanism behind `dag.if(...)`: a condition's boolean is a run-time
115-
* signal, so which branch a Dag run takes is decided here rather than
116-
* recorded in the Dag. Python reaches the same supervisor message through
117-
* `skip()` and `skip_all_except()`.
118-
*
119-
* Each id must name a direct downstream of the running task. Passing none is
120-
* a no-op, so a branch that skips nothing needs no guard at the call site.
121-
*/
122-
skipDownstreamTasks(taskIds: readonly string[]): Promise<void>;
123110
}
124111

125112
/** Error thrown by {@link TaskClient.getVariableOrThrow}. */

0 commit comments

Comments
 (0)