Skip to content

Commit 8fb2d5f

Browse files
committed
TS SDK: default a task id and argument labels to the handler's names
1 parent ea1f472 commit 8fb2d5f

7 files changed

Lines changed: 387 additions & 38 deletions

File tree

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

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -277,6 +277,18 @@ argument itself: one buried inside an array or an object is a literal, and draws
277277
Every task has to be called exactly once. An uncalled task fails when the Dag is read, so none can be
278278
left out of the graph by accident.
279279

280+
The task id may be omitted, in which case it is the handler's function name:
281+
282+
.. code-block:: typescript
283+
284+
const extract = dag.task(async function extract(): Promise<number> {
285+
return 42;
286+
});
287+
288+
``airflow-ts-pack`` keeps function names intact, so bundling cannot rename a task. A handler with no
289+
name of its own, such as an arrow function passed inline, has nothing to take an id from and needs
290+
one: either positionally or as ``taskId`` in its spec. Give it in one place only, not both.
291+
280292
``new Dag`` and ``dag.task`` both take a trailing spec of Airflow options:
281293
``{ schedule: "@daily", tags: ["etl"] }`` for the Dag, ``{ retries: 2, retryDelay: 30 }`` for a task.
282294

@@ -398,9 +410,9 @@ layout header. The layout records the byte ranges and SHA-256 digests of the man
398410
so there is one file to deploy, with no separate manifest or ``node_modules``.
399411

400412
The code is minified because an integrity digest is only worth taking over an artifact nobody is expected to
401-
read or edit in place. The ``/*! */`` license banners of bundled dependencies are kept. Nothing is identified by
402-
a function name, so minified names are safe: a Dag and a task are named by the string ids their registration
403-
states, and a handler is dispatched by reference.
413+
read or edit in place. Whitespace and syntax are minified, not identifiers: a task id defaults to its
414+
handler's name, so the names the SDK reads have to be the ones the author wrote. The ``/*! */`` license
415+
banners of bundled dependencies are kept.
404416

405417
Because the shipped code is not the code anyone wrote, the packer also embeds the entry module verbatim in a
406418
``/*# airflowSource ... #*/`` block comment, verified by its own digest, so Airflow has something readable to

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

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -149,9 +149,9 @@ convention.
149149
by design. Native declaration is what fills them, generated from the serialized-Dag JSON schema the
150150
way `src/generated/supervisor.ts` is. This ADR does not choose those fields; it fixes where an
151151
author writes them.
152-
- `TaskOptions` carries the spec and the handler's positional argument names, which the packer fills in
153-
from the parameter list so the Dag names each argument as its handler does. With wiring moved to the
154-
factory call, `inputs` is no longer an option.
152+
- `TaskOptions` carries the spec and the handler's positional argument names, which the Dag reads
153+
from the handler's parameter list so it names each argument as the handler does. With wiring moved
154+
to the factory call, `inputs` is no longer an option.
155155
- `TaskHandlerArgs` is removed from the public API, `DagRegistry` becomes `Bundle`, and
156156
`serveDags(registry)` becomes `bundle.serve()`, which breaks
157157
0.1.0-beta1 authors; see [ADR-0001](0001-mixed-lang-dag-interface.md) for the shipped call sites
@@ -182,16 +182,17 @@ convention.
182182
and runs `validateEmptySpec` on it (`ts-sdk/src/sdk/dag.ts`), so task fields land on a path that
183183
exists rather than a new one.
184184
- **A positional argument binds by order, and its name is a label.** The serialized Dag names each
185-
argument, so the packer reads the names from the handler's parameter list; `arg0`, `arg1` and so on
186-
stand in for a name it cannot see, without changing which value reaches which argument.
185+
argument, so the names come from the handler's parameter list; `arg0`, `arg1` and so on stand in
186+
for a parameter with no name to read, without changing which value reaches which argument.
187187
- **A `TaskRef` is inert** — a handle for wiring, not a promise. Nothing in a Dag file executes a task
188188
body.
189-
- **A defaulted task id is resolved at pack time, not read at runtime.** esbuild renames function
190-
identifiers, so `handler.name` in a packed bundle is the minified name, not the author's. The pack
191-
step (`ts-sdk/src/cli/pack.ts`) therefore reads an omitted id from the handler's declared name in
192-
source and writes it into the registration, rather than depending on `handler.name` or enabling
193-
esbuild's `keepNames` across the whole bundle. A handler with no source name leaves nothing to
194-
read, which is why an anonymous handler must state its id.
189+
- **A defaulted task id and the argument labels are read off the handler itself.** The pack step
190+
(`ts-sdk/src/cli/pack.ts`) minifies whitespace and syntax but not identifiers, and passes esbuild's
191+
`keepNames`, so `handler.name` and the handler's parameter names are the author's in a packed
192+
bundle as much as in one run from source. Rewriting the call at pack time was tried first and
193+
dropped: it needed a TypeScript parser in the packer to tell a real `.task(` from one inside a
194+
string or a comment, and it could not see a handler declared in another module. A handler with no
195+
name leaves nothing to read, which is why an anonymous handler must state its id.
195196
- **`withArgNames` and the name folding behind it** ([ADR-0001](0001-mixed-lang-dag-interface.md))
196197
exist for the mixed-language case and are never needed here: both ends of every name are
197198
TypeScript, so `tsc` checks the wiring end to end and there is no foreign name to reconcile.

‎ts-sdk/src/cli/pack.ts‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -215,8 +215,14 @@ export async function runPack(argv: readonly string[]): Promise<void> {
215215
platform: "node",
216216
format: "esm",
217217
target: "node22",
218-
// A digest is only worth taking over an artifact nobody reads or edits in place.
219-
minify: true,
218+
// Whitespace and syntax only. Identifiers stay: `dag.task(handler)` takes
219+
// the task id from the handler's name and labels its arguments after the
220+
// handler's parameters, and both are read off the function at run time.
221+
minifyWhitespace: true,
222+
minifySyntax: true,
223+
// Bundling renames a symbol that two modules both declare, so the name
224+
// the SDK reads is pinned to the one the author wrote.
225+
keepNames: true,
220226
// The manifest is read by running the staged bundle, so the metadata describes what ships.
221227
outfile: stagingPath,
222228
});

‎ts-sdk/src/sdk/dag.ts‎

Lines changed: 128 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,9 @@ function kindOf(value: object): string {
8484
}
8585

8686
const DAG_SPEC_KEYS: ReadonlySet<string> = new Set(Object.keys(DAG_SCHEMA_FIELDS));
87-
const TASK_SPEC_KEYS: ReadonlySet<string> = new Set(Object.keys(TASK_SCHEMA_FIELDS));
87+
// `taskId` is hand-written rather than generated: the schema's task_id is
88+
// serializer-owned, and this is the authoring surface's own way to set it.
89+
const TASK_SPEC_KEYS: ReadonlySet<string> = new Set([...Object.keys(TASK_SCHEMA_FIELDS), "taskId"]);
8890

8991
/**
9092
* Dag-level options: the schedule, the tags, how many runs may be active, and
@@ -103,9 +105,19 @@ export type DagSpec = GeneratedDagFields;
103105
* Task-level options: the retries, the pool, the trigger rule, and the rest of
104106
* what an operator takes in Python.
105107
*
108+
* The task id is here too, for a handler whose id is not given positionally.
106109
* Optional and record-only on the same terms as {@link DagSpec}.
107110
*/
108-
export type TaskSpec = GeneratedTaskFields;
111+
export interface TaskSpec extends GeneratedTaskFields {
112+
/**
113+
* Airflow task ID, when it should not be the handler's function name.
114+
*
115+
* `dag.task(handler)` takes the id from the handler's name. Set this for an
116+
* id that has to outlive that name, or for an anonymous handler, which has
117+
* no name to take one from.
118+
*/
119+
readonly taskId?: string;
120+
}
109121

110122
// Carries a reference's return type without carrying a value. Not exported, so
111123
// the property cannot be read or written from outside; it exists only so
@@ -259,10 +271,11 @@ export type TaskOptions = TaskSpec & {
259271
/**
260272
* Names for the handler's positional arguments, in declaration order.
261273
*
262-
* `airflow-ts-pack` fills this in from the handler's parameter list, so the
263-
* Dag names each argument as its handler does. Positional inputs bind by
264-
* order, so a name left out only costs the label: `arg0`, `arg1` and so on
265-
* stand in for it.
274+
* Read from the handler itself when it declares plain parameters, so the Dag
275+
* names each argument as its handler does. Set this to name them yourself,
276+
* or for a handler whose parameters are destructured, defaulted or rest.
277+
* Positional inputs bind by order, so a name left out only costs the label:
278+
* `arg0`, `arg1` and so on stand in for it.
266279
*/
267280
readonly argBindings?: readonly string[];
268281
};
@@ -354,12 +367,70 @@ export class Dag {
354367
* {@link TaskFactory}; `getContext()` and `getClient()` reach the runtime
355368
* from inside the call, so neither is an argument. The trailing options object
356369
* carries this task's own {@link TaskSpec}.
370+
*
371+
* ```ts
372+
* const extract = dag.task("extract", async () => 42);
373+
* const transform = dag.task(async function transform() {}); // id "transform"
374+
* ```
357375
*/
358376
task<TParams extends readonly unknown[] = [], TReturn = unknown>(
359377
taskId: string,
360378
handler: (...args: TParams) => TReturn | Promise<TReturn>,
361-
options: TaskOptions = {},
379+
options?: TaskOptions,
380+
): TaskFactory<TParams, TReturn>;
381+
/**
382+
* Declare a task whose id is the handler's function name.
383+
*
384+
* `airflow-ts-pack` keeps handler names intact, so minification cannot change
385+
* a task id. An anonymous handler has no name to take one from, and needs
386+
* {@link TaskSpec.taskId} to give it one.
387+
*/
388+
task<TParams extends readonly unknown[] = [], TReturn = unknown>(
389+
handler: (...args: TParams) => TReturn | Promise<TReturn>,
390+
options?: TaskOptions,
391+
): TaskFactory<TParams, TReturn>;
392+
task<TParams extends readonly unknown[] = [], TReturn = unknown>(
393+
taskIdOrHandler: string | ((...args: TParams) => TReturn | Promise<TReturn>),
394+
handlerOrOptions?: ((...args: TParams) => TReturn | Promise<TReturn>) | TaskOptions,
395+
maybeOptions?: TaskOptions,
362396
): TaskFactory<TParams, TReturn> {
397+
const idGiven = typeof taskIdOrHandler === "string";
398+
const handler = (idGiven ? handlerOrOptions : taskIdOrHandler) as (
399+
...args: TParams
400+
) => TReturn | Promise<TReturn>;
401+
const given = idGiven ? maybeOptions : (handlerOrOptions as TaskOptions | undefined);
402+
// Defaulted only when absent: an explicit `null` is a bad spec, not an
403+
// omitted one, and #taskSpecOf is what reports it.
404+
const options = given === undefined ? {} : given;
405+
const specTaskId =
406+
isPlainRecord(options) && typeof options.taskId === "string" ? options.taskId : undefined;
407+
// Two ids for one task disagree silently otherwise: the positional one
408+
// wins and the spec's is dropped without a word.
409+
if (idGiven && specTaskId !== undefined) {
410+
throw new Error(
411+
`Task "${taskIdOrHandler}" of Dag "${this.dagId}" also carries taskId "${specTaskId}" in ` +
412+
"its spec; give the id once, either positionally or in the spec",
413+
);
414+
}
415+
const defaulted = idGiven ? undefined : (specTaskId ?? readFunctionName(handler));
416+
const taskId = idGiven ? taskIdOrHandler : defaulted;
417+
if (taskId === undefined) {
418+
throw new Error(
419+
`A task of Dag "${this.dagId}" has no id: its handler has no name to take one from. ` +
420+
'Pass an id — dag.task("my_task", handler) — or give the handler a name. A bundler ' +
421+
"that drops function names also lands here; airflow-ts-pack keeps them.",
422+
);
423+
}
424+
// A name Airflow would reject is worth catching where it was taken, not in
425+
// the server's answer: `fn.bind(...)` names itself "bound extract", and a
426+
// method can be named anything at all.
427+
if (defaulted !== undefined && !TASK_ID_CHARACTERS.test(defaulted)) {
428+
throw new Error(
429+
`A task of Dag "${this.dagId}" would take the id "${defaulted}" from its handler's name, ` +
430+
"which Airflow does not accept; give it an id of letters, digits, dashes, dots and " +
431+
'underscores — dag.task("my_task", handler)',
432+
);
433+
}
363434
if (typeof handler !== "function") {
364435
throw new Error(`handler for Dag "${this.dagId}" task "${taskId}" must be a function`);
365436
}
@@ -375,7 +446,8 @@ export class Dag {
375446
);
376447
}
377448
const spec = this.#taskSpecOf(taskId, options);
378-
const argBindings = this.#validateArgBindings(taskId, options.argBindings);
449+
const argBindings =
450+
this.#validateArgBindings(taskId, options.argBindings) ?? inferArgBindings(handler);
379451
const task = createTaskRef(this.dagId, taskId);
380452
this.#tasks.set(taskId, {
381453
task,
@@ -560,6 +632,54 @@ function spreadArgs(fn: TaskFunction): TaskFunction {
560632
handler(...Object.values(args ?? {}))) as unknown as TaskFunction;
561633
}
562634

635+
// What Airflow accepts as a task id, and so what a name taken from a handler
636+
// has to look like.
637+
const TASK_ID_CHARACTERS = /^[\p{L}\p{N}_.-]+$/u;
638+
639+
// A parameter list of plain identifiers, which is the one shape whose names can
640+
// be read off a function without parsing it.
641+
const PLAIN_PARAMETERS = /^[A-Za-z_$][\w$]*(?:\s*,\s*[A-Za-z_$][\w$]*)*$/;
642+
643+
/**
644+
* A handler's own function name, or undefined when it has none.
645+
*
646+
* Where a defaulted task id comes from. `airflow-ts-pack` bundles with esbuild's
647+
* `keepNames`, so the name survives minification; a bundler that drops names
648+
* leaves the empty string, which is why the empty string is not an id.
649+
*/
650+
function readFunctionName(handler: unknown): string | undefined {
651+
if (typeof handler !== "function") return undefined;
652+
const { name } = handler as { name?: unknown };
653+
return typeof name === "string" && name.length > 0 ? name : undefined;
654+
}
655+
656+
/**
657+
* The handler's parameter names, when it declares plain ones.
658+
*
659+
* All-or-nothing: a default, a destructuring pattern, a rest element or a
660+
* comment leaves the labels `arg0`, `arg1`, which bind exactly the same way.
661+
* `airflow-ts-pack` leaves identifiers alone, so these are the names the author
662+
* wrote even in a packed bundle.
663+
*/
664+
function inferArgBindings(handler: unknown): readonly string[] | undefined {
665+
if (typeof handler !== "function") return undefined;
666+
const source = Function.prototype.toString.call(handler);
667+
// `async rows => ...`: one parameter, and no parentheses to slice.
668+
const bare = /^(?:async\s+)?([A-Za-z_$][\w$]*)\s*=>/.exec(source);
669+
if (bare) return Object.freeze([bare[1]!]);
670+
const open = source.indexOf("(");
671+
const close = source.indexOf(")");
672+
const arrow = source.indexOf("=>");
673+
// A "(" past the arrow belongs to the body, not to a parameter list.
674+
if (open === -1 || close < open || (arrow !== -1 && open > arrow)) return undefined;
675+
const parameters = source.slice(open + 1, close).trim();
676+
if (parameters === "" || !PLAIN_PARAMETERS.test(parameters)) return undefined;
677+
const names = parameters.split(",").map((name) => name.trim());
678+
// `length` stops at the first default or rest element, so it matches only a
679+
// plain list — and not a slice that a ")" inside a default value cut short.
680+
return names.length === handler.length ? Object.freeze(names) : undefined;
681+
}
682+
563683
function createTaskRef(dagId: string, taskId: string): TaskRef {
564684
const task: TaskRef = { dagId, taskId };
565685
brand(task, "TaskRef");

‎ts-sdk/tests/cli/pack.test.ts‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -557,6 +557,104 @@ describe("runPack", () => {
557557
);
558558
});
559559

560+
it("takes an omitted task id from the handler name, through minification", async () => {
561+
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
562+
const entry = path.join(outdir, "named-entry.ts");
563+
writeFileSync(
564+
entry,
565+
[
566+
`import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
567+
'const salesDag = new Dag("sales_dag");',
568+
"salesDag.task(async function extractRows() {})();",
569+
"async function loadRows() {}",
570+
"salesDag.task(loadRows, { retries: 2 })();",
571+
"await new Bundle(salesDag).serve();",
572+
].join("\n"),
573+
);
574+
const stderr = captureStderr();
575+
576+
await runPack([entry, "--outdir", outdir]);
577+
578+
// Read back off the packed artifact, so this asserts what minification
579+
// left behind rather than what the source said.
580+
expect(JSON.parse(readEmbeddedMetadata(path.join(outdir, "bundle.min.mjs")))).toHaveProperty(
581+
"task_handlers.sales_dag.tasks",
582+
["extractRows", "loadRows"],
583+
);
584+
expect(stderr()).toBe("");
585+
});
586+
587+
it("names a positional handler's arguments, through minification", async () => {
588+
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
589+
const entry = path.join(outdir, "positional-entry.ts");
590+
writeFileSync(
591+
entry,
592+
[
593+
`import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
594+
'const salesDag = new Dag("sales_dag");',
595+
"const transform = salesDag.task(async function transformRows(rows: number, region: string) {",
596+
" return `${region}:${rows}`;",
597+
"});",
598+
'transform(1, "us");',
599+
"await new Bundle(salesDag).serve();",
600+
].join("\n"),
601+
);
602+
603+
await runPack([entry, "--outdir", outdir]);
604+
605+
// The Dag reads the names off the handler when it registers, so they have
606+
// to survive packing as the parameters themselves.
607+
const bundle = readFileSync(path.join(outdir, "bundle.min.mjs"), "utf8");
608+
// minifySyntax drops a function expression's own name, which is why
609+
// `keepNames` carries the task id; the parameters are what stay verbatim.
610+
expect(bundle).toMatch(/async function\s*\(rows,\s*region\)/);
611+
expect(JSON.parse(readEmbeddedMetadata(path.join(outdir, "bundle.min.mjs")))).toHaveProperty(
612+
"task_handlers.sales_dag.tasks",
613+
["transformRows"],
614+
);
615+
});
616+
617+
it("fails the pack when a handler is anonymous and names no task", async () => {
618+
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
619+
const entry = path.join(outdir, "anonymous-entry.ts");
620+
writeFileSync(
621+
entry,
622+
[
623+
`import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
624+
'const salesDag = new Dag("sales_dag");',
625+
"salesDag.task(async () => undefined)();",
626+
"await new Bundle(salesDag).serve();",
627+
].join("\n"),
628+
);
629+
630+
await expect(runPack([entry, "--outdir", outdir])).rejects.toThrow(
631+
/has no id: its handler has no name/,
632+
);
633+
});
634+
635+
it("takes a computed task id as the id it evaluates to", async () => {
636+
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
637+
const entry = path.join(outdir, "computed-entry.ts");
638+
writeFileSync(
639+
entry,
640+
[
641+
`import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
642+
'const salesDag = new Dag("sales_dag");',
643+
'const region = "north";',
644+
"salesDag.task(`extract_${region}`, async () => undefined)();",
645+
'salesDag.task("load_".concat(region), async () => undefined)();',
646+
"await new Bundle(salesDag).serve();",
647+
].join("\n"),
648+
);
649+
650+
await runPack([entry, "--outdir", outdir]);
651+
652+
expect(JSON.parse(readEmbeddedMetadata(path.join(outdir, "bundle.min.mjs")))).toHaveProperty(
653+
"task_handlers.sales_dag.tasks",
654+
["extract_north", "load_north"],
655+
);
656+
});
657+
560658
it("packs only the Dags the served bundle holds", async () => {
561659
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
562660
const entry = path.join(outdir, "forgotten-entry.ts");

0 commit comments

Comments
 (0)