jason810496 commented on code in PR #75565:
URL: https://github.com/apache/airflow/pull/75565#discussion_r4505560935
##########
ts-sdk/src/sdk/dag.ts:
##########
@@ -1242,6 +1268,10 @@ export class Dag {
) => this.#addTask<TArgs, TReturn>(groupId, taskIdOrHandler,
handlerOrOptions, maybeOptions),
taskGroup: (childId: string, options?: TaskGroupOptions) =>
this.#addGroup(groupId, childId, options),
+ if: (handler: (args: never) => boolean | Promise<boolean>, ...args:
readonly unknown[]) =>
+ this.#placeCondition(`taskGroup("${groupId}").if`, groupId, handler,
args),
Review Comment:
The read-time hints in `#finalize` (lines 1509 and 1517, outside this diff)
still say `dag.if` / `dag.switch` for a group's condition or switch; neutral
hints fit both.
```diff
- "give it one with dag.if(handler, inputs).then(task)",
+ "give it one with .then(task)",
...
- "give it the tasks to choose between with dag.switch(handler,
inputs).case(task)",
+ "give it the tasks to choose between with .case(task)",
```
##########
ts-sdk/tests/sdk/conditional.test.ts:
##########
@@ -219,6 +220,19 @@ describe("dag.if", () => {
expect(getDagOrderEdges(dag)).toContainEqual({ upstream: "has_rows",
downstream: "notify" });
});
+ it("declares a condition in a task group, under the group's prefix", async
() => {
+ const dag = new Dag("d");
+ const ifReady = dag.task("load_if_ready", async () => undefined)();
+ const fallback = dag.task("load_fallback", async () => undefined)();
+ const gate = dag.taskGroup("checks").if(async () => true, undefined, {
taskId: "has_rows" });
+ gate.then(ifReady).else(fallback);
+
+ expect(gate.taskId).toBe("checks.has_rows");
+
expect(getDagTaskGroups(dag).get("checks")?.taskIds).toEqual(["checks.has_rows"]);
+ const { skipDownstreamTasks } = await runHandler(dag, "checks.has_rows");
+ expect(skipDownstreamTasks).toHaveBeenCalledWith(["load_fallback"]);
+ });
Review Comment:
The "was awaited" message now uses the group label too, but nothing covers
it.
```suggestion
});
it("names the group method when a group's condition is awaited", async ()
=> {
const gate = new Dag("d")
.taskGroup("checks")
.if(async () => true, undefined, { taskId: "has_rows" });
await expect(Promise.resolve(gate as unknown as
Promise<void>)).rejects.toThrow(
/taskGroup\("checks"\)\.if\(\.\.\.\) of Dag "d" was awaited/,
);
});
```
##########
ts-sdk/tests/sdk/trigger-dag-run.test.ts:
##########
@@ -250,6 +250,19 @@ describe("triggerDagRun", () => {
);
});
+ it.each(["if", "switch"] as const)("a trigger task given to a task group's
%s", (decider) => {
+ const group = new Dag("d").taskGroup("routing");
+ const declare = group[decider].bind(group) as (...args: unknown[]) =>
unknown;
Review Comment:
The group's members are arrow functions, so `.bind(group)` is not needed.
```suggestion
const declare = group[decider] as (...args: unknown[]) => unknown;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
