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 d3ddca0c9b49b..e40b1363c0ad7 100644 --- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst +++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst @@ -247,8 +247,8 @@ A ``Dag`` is declared on this side rather than in Python: its schedule, its task the edges between them are all written in TypeScript. The surface is still growing, so a Dag declared this way is not served to Airflow yet. -``dag.task(taskId, handler)`` returns a *factory*. Calling it places the task in the Dag and supplies the -handler's arguments, so the call graph is the task graph: +``dag.task(taskId, handler)`` returns a *factory*. A handler takes one object of named arguments, and +calling the factory names each input, so the call graph is the task graph: .. code-block:: typescript @@ -257,19 +257,19 @@ handler's arguments, so the call graph is the task graph: const dag = new Dag("ts_etl"); const extract = dag.task("extract", async (): Promise => 42); - const transform = dag.task("transform", async (rows: number, region: string) => rows * 2); - const load = dag.task("load", async (total: number) => {}); + const transform = dag.task( + "transform", + async ({ rows, region }: { rows: number; region: string }) => rows * 2, + ); + const load = dag.task("load", async ({ total }: { total: number }) => {}); - load(transform(extract(), "us")); + const extracted = extract(); + const total = transform({ rows: extracted, region: "us" }); + load({ total }); -Arguments are passed in the order the handler declares them. A handler that declares a single object of -named arguments can also be called with that object, which names each input instead of ordering it: - -.. code-block:: typescript - - const store = dag.task("store", async ({ total }: { total: number }) => {}); - - store({ total: extract() }); +Naming the inputs is how a task is called. A handler that takes no arguments is called with none, and +a single argument is named like any other, ``load({ total })``. The compiler checks the call: it reports +an argument left out, a misspelled one, and a literal of the wrong type. Each argument takes either an upstream reference or a literal JSON value. A reference has to be the argument itself: one buried inside an array or an object is a literal, and draws no edge. @@ -277,6 +277,18 @@ argument itself: one buried inside an array or an object is a literal, and draws Every task has to be called exactly once. An uncalled task fails when the Dag is read, so none can be left out of the graph by accident. +The task id may be omitted, in which case it is the handler's function name: + +.. code-block:: typescript + + const extract = dag.task(async function extract(): Promise { + return 42; + }); + +``airflow-ts-pack`` keeps function names intact, so bundling cannot rename a task. A handler with no +name of its own, such as an arrow function passed inline, has nothing to take an id from and needs +one: either positionally or as ``taskId`` in its spec. Give it in one place only, not both. + ``new Dag`` and ``dag.task`` both take a trailing spec of Airflow options: ``{ schedule: "@daily", tags: ["etl"] }`` for the Dag, ``{ retries: 2, retryDelay: 30 }`` for a task. @@ -398,9 +410,8 @@ layout header. The layout records the byte ranges and SHA-256 digests of the man so there is one file to deploy, with no separate manifest or ``node_modules``. The code is minified because an integrity digest is only worth taking over an artifact nobody is expected to -read or edit in place. The ``/*! */`` license banners of bundled dependencies are kept. Nothing is identified by -a function name, so minified names are safe: a Dag and a task are named by the string ids their registration -states, and a handler is dispatched by reference. +read or edit in place. Function names are kept through minification, since a task id defaults to its +handler's name. The ``/*! */`` license banners of bundled dependencies are kept. Because the shipped code is not the code anyone wrote, the packer also embeds the entry module verbatim in a ``/*# airflowSource ... #*/`` block comment, verified by its own digest, so Airflow has something readable to diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt index 03876053a19a8..2f15cb46aefa0 100644 --- a/docs/spelling_wordlist.txt +++ b/docs/spelling_wordlist.txt @@ -1100,6 +1100,7 @@ midnights milli millis milton +minification minikube misconfiguration misconfigured diff --git a/ts-sdk/adr/0002-native-dag-interface.md b/ts-sdk/adr/0002-native-dag-interface.md index eca1f5e2ea294..6614ee62091bb 100644 --- a/ts-sdk/adr/0002-native-dag-interface.md +++ b/ts-sdk/adr/0002-native-dag-interface.md @@ -27,16 +27,13 @@ Proposed. Revised after the review on #72047. 1. **`dag.task(handler)` returns a factory, and the task id is optional.** With no id the task takes the handler's function name (`dag.task(extract)` → task `"extract"`); `dag.task(taskId, handler)` - sets it explicitly, which an anonymous handler must do. Calling the factory both places the task in - the Dag and supplies its arguments, in the order the handler declares them - (`load(transform(extract(), "us"))`). A handler that declares a single object of named arguments can - also be called with that object — the shape Python TaskFlow uses for - `load(transformed=transform(...))`. -2. **The call graph is the task graph.** `tsc` checks every wired key against the handler's own - parameter type, and a `TaskRef` exists only once its producing call has returned, so a cycle - through arguments is unrepresentable rather than rejected by a validator. A reference passed by - position is checked against the argument's own type, which is what tells the two call shapes apart - when a handler declares a single argument. + sets it explicitly, which an anonymous handler must do. A handler takes one object of named + arguments, and calling the factory both places the task in the Dag and names each of its inputs + (`load({ total: transform({ rows: extract(), region: "us" }) })`) — the shape Python TaskFlow uses + for `load(transformed=transform(...))`. +2. **The call graph is the task graph.** `tsc` checks every named input against the handler's own + argument type, and a `TaskRef` exists only once its producing call has returned, so a cycle + through arguments is unrepresentable rather than rejected by a validator. 3. **Every task is called exactly once.** An uncalled task fails when the Dag is read, so none can be silently left out of the graph. 4. **`before` and `after` draw order-only edges** — the TypeScript pair for `>>` and `<<`, both @@ -107,13 +104,12 @@ const extract = dag.task(async function extract(): Promise { // task id "extract" ``` -The id comes from the handler's *source* name, resolved when the bundle is packed and written into -the registration — not from `handler.name` at runtime, which minification renames (see -Implementation Notes). A handler with no source name — a bare anonymous arrow passed inline, -`dag.task(async () => 42)` — has nothing to resolve and is a compile error until given an explicit -id. This default is for native Dags, where both ends of every name are TypeScript; a mixed-language -handler names the Python-owned task explicitly and does not default from the handler's function -name ([ADR-0001](0001-mixed-lang-dag-interface.md), decision 3). +The id is the handler's `name`, which `airflow-ts-pack` keeps through minification (see +Implementation Notes). A handler with no name, such as a bare anonymous arrow passed inline, +`dag.task(async () => 42)`, has nothing to take an id from and fails when the Dag is declared until +given an explicit id. This default is for native Dags, where both ends of every name are +TypeScript; a mixed-language handler names the Python-owned task explicitly and does not default +from the handler's function name ([ADR-0001](0001-mixed-lang-dag-interface.md), decision 3). The `TaskSpec` also carries the task id, so it can be set alongside the other task options: @@ -149,9 +145,8 @@ convention. by design. Native declaration is what fills them, generated from the serialized-Dag JSON schema the way `src/generated/supervisor.ts` is. This ADR does not choose those fields; it fixes where an author writes them. -- `TaskOptions` carries the spec and the handler's positional argument names, which the packer fills in - from the parameter list so the Dag names each argument as its handler does. With wiring moved to the - factory call, `inputs` is no longer an option. +- `TaskOptions` carries the task's spec and nothing else: the names on the wire are the keys of the + call itself. With wiring moved to the factory call, `inputs` is no longer an option. - `TaskHandlerArgs` is removed from the public API, `DagRegistry` becomes `Bundle`, and `serveDags(registry)` becomes `bundle.serve()`, which breaks 0.1.0-beta1 authors; see [ADR-0001](0001-mixed-lang-dag-interface.md) for the shipped call sites @@ -159,10 +154,10 @@ convention. ## Alternatives -- **Named-only wiring**, rejected in the review on #73435: naming every input reads well at twenty - tasks but forces an object around a single argument, and positional calls are what TypeScript - authors write. Both are offered, and the handler's own parameter list decides which one a task can - use. +- **Positional handlers**, `async (rows: number, region: string) => ...`, offered first and then + dropped: a positional parameter list has no names on the wire unless the SDK reads them out of the + handler's source, and a single object of named arguments is what a TypeScript library takes + anyway. - **Injected `ctx`/`client` arguments**, mimicking the Python signature. Rejected, per the above and because feeling native to TypeScript matters more than matching Python's parameter list. @@ -181,17 +176,14 @@ convention. - **The spec argument already has its slot.** `dag.task(taskId, handler, options)` reads `{ spec = {} }` and runs `validateEmptySpec` on it (`ts-sdk/src/sdk/dag.ts`), so task fields land on a path that exists rather than a new one. -- **A positional argument binds by order, and its name is a label.** The serialized Dag names each - argument, so the packer reads the names from the handler's parameter list; `arg0`, `arg1` and so on - stand in for a name it cannot see, without changing which value reaches which argument. - **A `TaskRef` is inert** — a handle for wiring, not a promise. Nothing in a Dag file executes a task body. -- **A defaulted task id is resolved at pack time, not read at runtime.** esbuild renames function - identifiers, so `handler.name` in a packed bundle is the minified name, not the author's. The pack - step (`ts-sdk/src/cli/pack.ts`) therefore reads an omitted id from the handler's declared name in - source and writes it into the registration, rather than depending on `handler.name` or enabling - esbuild's `keepNames` across the whole bundle. A handler with no source name leaves nothing to - read, which is why an anonymous handler must state its id. +- **A defaulted task id is read off the handler itself.** The pack step (`ts-sdk/src/cli/pack.ts`) + minifies and passes esbuild's `keepNames`, so `handler.name` is the author's in a packed bundle as + much as in one run from source. Rewriting the call at pack time was tried first and dropped: it + needed a TypeScript parser in the packer to tell a real `.task(` from one inside a string or a + comment, and it could not see a handler declared in another module. A handler with no name leaves + nothing to read, which is why an anonymous handler must state its id. - **`withArgNames` and the name folding behind it** ([ADR-0001](0001-mixed-lang-dag-interface.md)) exist for the mixed-language case and are never needed here: both ends of every name are TypeScript, so `tsc` checks the wiring end to end and there is no foreign name to reconcile. diff --git a/ts-sdk/src/cli/pack.ts b/ts-sdk/src/cli/pack.ts index 10e8c5fbbdc8a..3d33d3b00411f 100644 --- a/ts-sdk/src/cli/pack.ts +++ b/ts-sdk/src/cli/pack.ts @@ -217,6 +217,10 @@ export async function runPack(argv: readonly string[]): Promise { target: "node22", // A digest is only worth taking over an artifact nobody reads or edits in place. minify: true, + // A task id comes from the handler's name, and bundling renames a symbol + // that two modules both declare, so the name the SDK reads is pinned to + // the one the author wrote. + keepNames: true, // The manifest is read by running the staged bundle, so the metadata describes what ships. outfile: stagingPath, }); diff --git a/ts-sdk/src/index.ts b/ts-sdk/src/index.ts index d63bdf24186c8..ad45c3bad336c 100644 --- a/ts-sdk/src/index.ts +++ b/ts-sdk/src/index.ts @@ -28,7 +28,6 @@ export type { ArgNameMap } from "./sdk/arg-names.js"; export type { Registerable } from "./sdk/bundle.js"; export type { DagSpec, - PositionalInputs, TaskFactory, TaskInput, TaskInputs, diff --git a/ts-sdk/src/sdk/dag.ts b/ts-sdk/src/sdk/dag.ts index a264d3061041d..2903b2b6ac4db 100644 --- a/ts-sdk/src/sdk/dag.ts +++ b/ts-sdk/src/sdk/dag.ts @@ -84,7 +84,9 @@ function kindOf(value: object): string { } const DAG_SPEC_KEYS: ReadonlySet = new Set(Object.keys(DAG_SCHEMA_FIELDS)); -const TASK_SPEC_KEYS: ReadonlySet = new Set(Object.keys(TASK_SCHEMA_FIELDS)); +// `taskId` is hand-written rather than generated: the schema's task_id is +// serializer-owned, and this is the authoring surface's own way to set it. +const TASK_SPEC_KEYS: ReadonlySet = new Set([...Object.keys(TASK_SCHEMA_FIELDS), "taskId"]); /** * Dag-level options: the schedule, the tags, how many runs may be active, and @@ -103,9 +105,19 @@ export type DagSpec = GeneratedDagFields; * Task-level options: the retries, the pool, the trigger rule, and the rest of * what an operator takes in Python. * + * The task id is here too, for a handler whose id is not given positionally. * Optional and record-only on the same terms as {@link DagSpec}. */ -export type TaskSpec = GeneratedTaskFields; +export interface TaskSpec extends GeneratedTaskFields { + /** + * Airflow task ID, when it should not be the handler's function name. + * + * `dag.task(handler)` takes the id from the handler's name. Set this for an + * id that has to outlive that name, or for an anonymous handler, which has + * no name to take one from. + */ + readonly taskId?: string; +} // Carries a reference's return type without carrying a value. Not exported, so // the property cannot be read or written from outside; it exists only so @@ -197,75 +209,46 @@ type JsonCompatible = T extends JsonValue */ export type TaskInput = TaskRef | JsonCompatible; -/** The inputs of a task that declares several arguments, in declaration order. */ -export type PositionalInputs = { - [K in keyof TParams]: TaskInput; -}; - -/** The inputs of a task that declares one object of named arguments, by name. */ +/** The inputs of a task, keyed by the name of the argument each one supplies. */ export type TaskInputs = { [K in keyof TArgs]: TaskRef | JsonCompatible; }; -// Offered only where it means something: `TaskInputs` would map over -// `number`'s own methods and accept `{ toFixed: ... }`. -type NamedInputs = [TOnly] extends [object] ? TaskInputs : never; - /** * What `dag.task(...)` returns: call it to declare where the task sits in the Dag. * - * Pass one input per argument the handler declares, in order. An input is - * either another task's reference, which makes this task wait for that task and - * receive its result, or a literal value: + * A handler takes one object of named arguments, and the call names each + * input. An input is either another task's reference, which makes this task + * wait for that task and receive its result, or a literal value: * * ```ts * const extract = dag.task("extract", async (): Promise => 42); - * const transform = dag.task("transform", async (rows: number, region: string) => rows); - * const load = dag.task("load", async (total: number) => {}); + * const transform = dag.task( + * "transform", + * async ({ rows, region }: { rows: number; region: string }) => rows, + * ); + * const load = dag.task("load", async ({ total }: { total: number }) => {}); * - * load(transform(extract(), "us")); + * const extracted = extract(); + * load({ total: transform({ rows: extracted, region: "us" }) }); * ``` * - * A handler that declares a single object of named arguments can also be called - * with that object, which names each input instead of ordering it: - * - * ```ts - * const store = dag.task("store", async ({ total }: { total: number }) => {}); - * - * store({ total: extract() }); - * ``` - * - * The compiler checks that every argument is supplied and that each literal - * matches its argument's type. A reference passed by position is checked against - * the argument's type as well, which is what tells the two call shapes apart - * when a handler declares a single argument. + * The call is checked argument by argument: every one has to be supplied, and + * a literal has to match its argument's type. */ -export type TaskFactory = [TParams] extends [ - readonly [], +export type TaskFactory = [TArgs] extends [ + void, ] ? () => TaskRef - : TParams extends readonly [infer TOnly] - ? (input: TaskInput | NamedInputs) => TaskRef - : (...inputs: PositionalInputs) => TaskRef; + : (inputs: TaskInputs) => TaskRef; /** - * The trailing argument of `dag.task()`: the task's own {@link TaskSpec}, plus - * the names of the handler's positional arguments. + * The trailing argument of `dag.task()`: the task's own {@link TaskSpec}. * * Every field is optional, and an unknown key is rejected, so a misspelled * field is an error rather than a task that quietly ignores it. */ -export type TaskOptions = TaskSpec & { - /** - * Names for the handler's positional arguments, in declaration order. - * - * `airflow-ts-pack` fills this in from the handler's parameter list, so the - * Dag names each argument as its handler does. Positional inputs bind by - * order, so a name left out only costs the label: `arg0`, `arg1` and so on - * stand in for it. - */ - readonly argBindings?: readonly string[]; -}; +export type TaskOptions = TaskSpec; /** Per-task record a Dag retains: the reference, the handler, and its spec. */ export interface TaskRecord { @@ -350,19 +333,86 @@ export class Dag { /** * Declare a task of this Dag, and return the factory that places it. * - * Every argument the handler declares becomes an input of the returned - * {@link TaskFactory}; `getContext()` and `getClient()` reach the runtime - * from inside the call, so neither is an argument. The trailing options object - * carries this task's own {@link TaskSpec}. + * A handler takes one object of named arguments, and every argument in it + * becomes an input of the returned {@link TaskFactory}; `getContext()` and + * `getClient()` reach the runtime from inside the call, so neither is an + * argument. The trailing options object carries this task's own + * {@link TaskSpec}. + * + * ```ts + * const extract = dag.task("extract", async () => 42); + * const transform = dag.task(async function transform() {}); // id "transform" + * ``` */ - task( + task( taskId: string, - handler: (...args: TParams) => TReturn | Promise, - options: TaskOptions = {}, - ): TaskFactory { + handler: (args: TArgs) => TReturn | Promise, + options?: TaskOptions, + ): TaskFactory; + /** + * Declare a task whose id is the handler's function name. + * + * `airflow-ts-pack` keeps handler names intact, so minification cannot change + * a task id. An anonymous handler has no name to take one from, and needs + * {@link TaskSpec.taskId} to give it one. + */ + task( + handler: (args: TArgs) => TReturn | Promise, + options?: TaskOptions, + ): TaskFactory; + task( + taskIdOrHandler: string | ((args: TArgs) => TReturn | Promise), + handlerOrOptions?: ((args: TArgs) => TReturn | Promise) | TaskOptions, + maybeOptions?: TaskOptions, + ): TaskFactory { + const idGiven = typeof taskIdOrHandler === "string"; + const handler = (idGiven ? handlerOrOptions : taskIdOrHandler) as ( + args: TArgs, + ) => TReturn | Promise; + const given = idGiven ? maybeOptions : (handlerOrOptions as TaskOptions | undefined); + // Defaulted only when absent: an explicit `null` is a bad spec, not an + // omitted one, and #taskSpecOf is what reports it. + const options = given === undefined ? {} : given; + const specTaskId = + isPlainRecord(options) && typeof options.taskId === "string" ? options.taskId : undefined; + // Two ids for one task disagree silently otherwise: the positional one + // wins and the spec's is dropped without a word. + if (idGiven && specTaskId !== undefined) { + throw new Error( + `Task "${taskIdOrHandler}" of Dag "${this.dagId}" also carries taskId "${specTaskId}" in ` + + "its spec; give the id once, either positionally or in the spec", + ); + } + const defaulted = idGiven ? undefined : (specTaskId ?? readFunctionName(handler)); + const taskId = idGiven ? taskIdOrHandler : defaulted; + if (taskId === undefined) { + throw new Error( + `A task of Dag "${this.dagId}" has no id: its handler has no name to take one from. ` + + 'Pass an id — dag.task("my_task", handler) — or give the handler a name. A bundler ' + + "that drops function names also lands here; airflow-ts-pack keeps them.", + ); + } + // A name Airflow would reject is worth catching where it was taken, not in + // the server's answer: `fn.bind(...)` names itself "bound extract", and a + // method can be named anything at all. + if (defaulted !== undefined && !TASK_ID_CHARACTERS.test(defaulted)) { + throw new Error( + `A task of Dag "${this.dagId}" would take the id "${defaulted}" from its handler's name, ` + + "which Airflow does not accept; give it an id of letters, digits, dashes, dots and " + + 'underscores — dag.task("my_task", handler)', + ); + } if (typeof handler !== "function") { throw new Error(`handler for Dag "${this.dagId}" task "${taskId}" must be a function`); } + // TypeScript already says so, but a plain-JavaScript author lands here + // with the argument list Python would take. + if (handler.length > 1) { + throw new Error( + `Handler for Dag "${this.dagId}" task "${taskId}" declares ${handler.length} parameters; ` + + "a handler takes one object of named arguments — async ({ rows, region }) => ...", + ); + } if (this.#tasks.has(taskId)) { throw new Error(`Task "${taskId}" is already registered for Dag "${this.dagId}"`); } @@ -375,25 +425,30 @@ export class Dag { ); } const spec = this.#taskSpecOf(taskId, options); - const argBindings = this.#validateArgBindings(taskId, options.argBindings); const task = createTaskRef(this.dagId, taskId); this.#tasks.set(taskId, { task, // The runtime dispatches every handler through one instantiation, as it - // does a registered TaskHandler; a positional one is wrapped at wiring. + // does a registered TaskHandler. fn: handler as unknown as TaskFunction, spec: freezeSpec(spec, () => `The spec for Dag "${this.dagId}" task "${taskId}"`), }); return ((...inputs: unknown[]) => { - this.#wire(taskId, inputs, argBindings); + // TypeScript already says so, but from plain JavaScript a second + // argument would be dropped without a word. + if (inputs.length > 1) { + throw new Error( + `Task "${taskId}" of Dag "${this.dagId}" was given ${inputs.length} arguments; ` + + "it takes one object naming its inputs: myTask({ rows, region })", + ); + } + this.#wire(taskId, inputs[0]); return task; - }) as TaskFactory; + }) as TaskFactory; } // TypeScript is bypassable — from plain JavaScript, or an `as TaskSpec` cast - // — so an unknown key is rejected rather than silently ignored. `argBindings` - // names the handler's arguments rather than configuring the task, so it is - // taken out here instead of reaching the spec. + // — so an unknown key is rejected rather than silently ignored. #taskSpecOf(taskId: string, options: TaskOptions): TaskSpec { const value: unknown = options; if (!isPlainRecord(value)) { @@ -401,7 +456,6 @@ export class Dag { } const spec: Record = {}; for (const key of Reflect.ownKeys(value)) { - if (key === "argBindings") continue; if (typeof key !== "string" || !TASK_SPEC_KEYS.has(key)) { throw new Error( `Unknown option "${String(key)}" in the spec for Dag "${this.dagId}" task "${taskId}"`, @@ -412,31 +466,7 @@ export class Dag { return spec as TaskSpec; } - // TypeScript is bypassable, and these names become the keys the arguments are - // recorded under, so an integer-like one would reorder what it labels. - #validateArgBindings(taskId: string, names: unknown): readonly string[] | undefined { - if (names === undefined) return undefined; - const describe = `argBindings for Dag "${this.dagId}" task "${taskId}"`; - if (!Array.isArray(names)) throw new Error(`${describe} must be an array of names`); - const seen = new Set(); - for (const name of names as unknown[]) { - if (typeof name !== "string" || name.length === 0 || /^\d+$/.test(name)) { - throw new Error( - `${describe} holds ${JSON.stringify(name)}; each name must be a non-empty ` + - "string that is not a number", - ); - } - if (seen.has(name)) throw new Error(`${describe} names "${name}" twice`); - seen.add(name); - } - return Object.freeze([...(names as string[])]); - } - - #wire( - taskId: string, - inputs: readonly unknown[], - argBindings: readonly string[] | undefined, - ): void { + #wire(taskId: string, inputs: unknown): void { if (this.#finalized) { throw new Error( `Task "${taskId}" of Dag "${this.dagId}" was called after the Dag was read; ` + @@ -449,16 +479,20 @@ export class Dag { "in a Dag, so call it once and reuse the reference", ); } - const positional = !isNamedCall(inputs); - const recorded = this.#checkInputs( - taskId, - positional ? positionalInputs(inputs, argBindings) : (inputs[0] as Record), - ); - this.#inputs.set(taskId, recorded); - if (positional && inputs.length > 0) { - const record = this.#tasks.get(taskId)!; - this.#tasks.set(taskId, { ...record, fn: spreadArgs(record.fn) }); + this.#inputs.set(taskId, this.#checkInputs(taskId, this.#inputsByName(taskId, inputs))); + } + + #inputsByName(taskId: string, inputs: unknown): Record { + if (inputs === undefined) return {}; + // A reference is a plain object too, so `load(extracted)` would otherwise + // read as a map of argument names. + if (!isPlainRecord(inputs) || isTaskRef(inputs)) { + throw new Error( + `Task "${taskId}" of Dag "${this.dagId}" takes one object naming its inputs: ` + + "myTask({ rows, region })", + ); } + return inputs; } #checkInputs(taskId: string, inputs: Record): RecordedInputs { @@ -525,39 +559,21 @@ export class Dag { } } -/** - * Whether a call named its inputs rather than ordering them. - * - * One plain object is the named form. A reference is a plain object too, so it - * is ruled out first: `load(extract())` is one positional input, not a map of - * argument names. - */ -function isNamedCall(inputs: readonly unknown[]): boolean { - return inputs.length === 1 && !isTaskRef(inputs[0]) && isPlainRecord(inputs[0]); -} - -function positionalInputs( - inputs: readonly unknown[], - argBindings: readonly string[] | undefined, -): Record { - const byName: Record = {}; - inputs.forEach((value, index) => { - byName[argBindings?.[index] ?? `arg${index}`] = value; - }); - return byName; -} +// What Airflow accepts as a task id, and so what a name taken from a handler +// has to look like. +const TASK_ID_CHARACTERS = /^[\p{L}\p{N}_.-]+$/u; /** - * Dispatch a positional handler through the one call shape the runtime uses. + * A handler's own function name, or undefined when it has none. * - * A task is called with its bound arguments as a single object, in the order - * they were recorded, so spreading its values back restores the argument list - * the handler declared. + * Where a defaulted task id comes from. `airflow-ts-pack` bundles with esbuild's + * `keepNames`, so the name survives minification; a bundler that drops names + * leaves the empty string, which is why the empty string is not an id. */ -function spreadArgs(fn: TaskFunction): TaskFunction { - const handler = fn as unknown as (...args: unknown[]) => unknown; - return ((args: Record) => - handler(...Object.values(args ?? {}))) as unknown as TaskFunction; +function readFunctionName(handler: unknown): string | undefined { + if (typeof handler !== "function") return undefined; + const { name } = handler as { name?: unknown }; + return typeof name === "string" && name.length > 0 ? name : undefined; } function createTaskRef(dagId: string, taskId: string): TaskRef { diff --git a/ts-sdk/tests/cli/pack.test.ts b/ts-sdk/tests/cli/pack.test.ts index c9d8e6a9ea63d..2ce54327cc898 100644 --- a/ts-sdk/tests/cli/pack.test.ts +++ b/ts-sdk/tests/cli/pack.test.ts @@ -557,6 +557,74 @@ describe("runPack", () => { ); }); + it("takes an omitted task id from the handler name, through minification", async () => { + outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-")); + const entry = path.join(outdir, "named-entry.ts"); + writeFileSync( + entry, + [ + `import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`, + 'const salesDag = new Dag("sales_dag");', + "salesDag.task(async function extractRows() {})();", + "async function loadRows() {}", + "salesDag.task(loadRows, { retries: 2 })();", + "await new Bundle(salesDag).serve();", + ].join("\n"), + ); + const stderr = captureStderr(); + + await runPack([entry, "--outdir", outdir]); + + // Read back off the packed artifact, so this asserts what minification + // left behind rather than what the source said. + expect(JSON.parse(readEmbeddedMetadata(path.join(outdir, "bundle.min.mjs")))).toHaveProperty( + "task_handlers.sales_dag.tasks", + ["extractRows", "loadRows"], + ); + expect(stderr()).toBe(""); + }); + + it("fails the pack when a handler is anonymous and names no task", async () => { + outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-")); + const entry = path.join(outdir, "anonymous-entry.ts"); + writeFileSync( + entry, + [ + `import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`, + 'const salesDag = new Dag("sales_dag");', + "salesDag.task(async () => undefined)();", + "await new Bundle(salesDag).serve();", + ].join("\n"), + ); + + await expect(runPack([entry, "--outdir", outdir])).rejects.toThrow( + /has no id: its handler has no name/, + ); + }); + + it("takes a computed task id as the id it evaluates to", async () => { + outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-")); + const entry = path.join(outdir, "computed-entry.ts"); + writeFileSync( + entry, + [ + `import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`, + 'const salesDag = new Dag("sales_dag");', + 'const region = "north";', + "salesDag.task(`extract_${region}`, async () => undefined)();", + 'salesDag.task("load_".concat(region), async () => undefined)();', + "await new Bundle(salesDag).serve();", + ].join("\n"), + ); + + await runPack([entry, "--outdir", outdir]); + + expect(JSON.parse(readEmbeddedMetadata(path.join(outdir, "bundle.min.mjs")))).toHaveProperty( + "task_handlers.sales_dag.tasks", + ["extract_north", "load_north"], + ); + }); + it("packs only the Dags the served bundle holds", async () => { outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-")); const entry = path.join(outdir, "forgotten-entry.ts"); diff --git a/ts-sdk/tests/public-api.test.ts b/ts-sdk/tests/public-api.test.ts index b3a55e54f585f..3ffba6332f999 100644 --- a/ts-sdk/tests/public-api.test.ts +++ b/ts-sdk/tests/public-api.test.ts @@ -27,7 +27,6 @@ import type { SetXComOpts, TaskClient, Registerable, - PositionalInputs, TaskContext, TaskFactory, TaskFunction, @@ -291,33 +290,33 @@ describe("public API", () => { // a wider one is, and not the other way round. expectTypeOf>().toMatchTypeOf(); expectTypeOf().not.toMatchTypeOf>(); - // Wiring moved to the factory call and the spec is the trailing argument - // itself, alongside the argument names the packer fills in. - expectTypeOf().toEqualTypeOf< - TaskSpec & { readonly argBindings?: readonly string[] } - >(); - // Each named argument takes any upstream reference or a literal of its own type. + // Wiring moved to the factory call, and the spec is the trailing argument + // itself. + expectTypeOf().toEqualTypeOf(); + // Each input takes any upstream reference or a literal of its own type. expectTypeOf>().toEqualTypeOf<{ rows: TaskRef | number }>(); - // A positional argument takes a literal or a reference of the argument's own - // type, which is what tells a one-argument positional call from a named one. expectTypeOf>().toEqualTypeOf | number>(); - expectTypeOf>().toEqualTypeOf< - [TaskRef | number, TaskRef | string] - >(); - // A handler with no arguments is called with none; one with several is - // called with a value per argument, in order. - expectTypeOf>().toEqualTypeOf<() => TaskRef>(); - expectTypeOf>().toEqualTypeOf<() => TaskRef>(); - expectTypeOf>().toEqualTypeOf< - (...inputs: [TaskRef | number, TaskRef | string]) => TaskRef + // A handler with no arguments is called with none; one that takes an object + // is called with its inputs named. + expectTypeOf().toEqualTypeOf<() => TaskRef>(); + expectTypeOf>().toEqualTypeOf<() => TaskRef>(); + expectTypeOf>().toEqualTypeOf< + (inputs: { rows: TaskRef | number }) => TaskRef >(); expectTypeOf>().toEqualTypeOf<[string, DagSpec?]>(); - expectTypeOf().toEqualTypeOf< - ( - taskId: string, - handler: (...args: TParams) => TReturn | Promise, - options?: TaskOptions, - ) => TaskFactory + // Overloaded, so the two forms are checked by calling them rather than by + // matching one signature. + const overloadedDag = new Dag("overload_dag"); + const withId = overloadedDag.task("extract", async (_: { rows: number }) => undefined); + const withoutId = overloadedDag.task(async function transform(_: { rows: number }) {}); + expectTypeOf(withId).toEqualTypeOf>(); + expectTypeOf(withoutId).toEqualTypeOf>(); + // A handler's return type reaches the reference its factory hands back. + expectTypeOf(overloadedDag.task("decide", async () => true)).toEqualTypeOf< + TaskFactory + >(); + expectTypeOf(overloadedDag.task("plain", async () => undefined)).toEqualTypeOf< + TaskFactory >(); expectTypeOf().toEqualTypeOf(); // Both specs are all-optional, so `{}` stays assignable and a field the @@ -333,9 +332,10 @@ describe("public API", () => { // still a plain optional: `undefined` already means "unset". expectTypeOf().toEqualTypeOf(); expectTypeOf().toEqualTypeOf(); - // Identity is positional, so it is not restated in either spec. + // Dag identity is positional, so the spec does not restate it. A task's is + // not: it is also what an anonymous handler sets instead of being named. expectTypeOf().not.toHaveProperty("dagId"); - expectTypeOf().not.toHaveProperty("taskId"); + expectTypeOf().toEqualTypeOf(); }); it("uses idiomatic TypeScript names for public client types", () => { @@ -423,7 +423,7 @@ describe("public API", () => { const rejectsPositionalMisuse = () => { // @ts-expect-error dagId is positional, not an options object. new Dag({ dagId: "example" }); - // @ts-expect-error a task handler is required. + // @ts-expect-error a task handler is required alongside an explicit id. new Dag("example").task("extract"); const dag = new Dag("example"); const extract = dag.task("extract", async () => undefined); @@ -449,17 +449,22 @@ describe("public API", () => { // @ts-expect-error a literal has to match its argument's type. transform({ rows: "many" }); const totals = dag.task("totals", async (): Promise<{ rows: number }> => ({ rows: 1 })); - // A single object of named arguments is given by name or by position. + // Inputs are named. transform({ rows: 1 }); + transform({ rows: totals() }); + // @ts-expect-error a bare reference is not an object of named inputs. transform(totals()); - // @ts-expect-error a positional reference has to return the argument's type. - transform(extract()); - const pair = dag.task("pair", async (rows: number, region: string) => `${region}${rows}`); + const pair = dag.task( + "pair", + async ({ rows, region }: { rows: number; region: string }) => `${region}${rows}`, + ); + pair({ rows: 1, region: "us" }); + // @ts-expect-error inputs are named, not given in order. pair(1, "us"); - // @ts-expect-error a positional argument cannot be skipped. - pair(1); - // @ts-expect-error each positional literal has to match its own argument. - pair("many", "us"); + // @ts-expect-error an argument cannot be skipped. + pair({ rows: 1 }); + // @ts-expect-error each literal has to match its own argument. + pair({ rows: "many", region: "us" }); const stamp = dag.task("stamp", async (_: { at: Date }) => undefined); // @ts-expect-error a Date cannot survive the serialized Dag, so only a reference will do. stamp({ at: new Date() }); diff --git a/ts-sdk/tests/sdk/dag.test.ts b/ts-sdk/tests/sdk/dag.test.ts index 098a2064b21d4..1bd493b602d7c 100644 --- a/ts-sdk/tests/sdk/dag.test.ts +++ b/ts-sdk/tests/sdk/dag.test.ts @@ -193,103 +193,45 @@ describe("Dag", () => { ); }); - it("records a positional call in argument order", () => { - const dag = new Dag("example_dag"); - const extract = dag.task("extract", async (): Promise => 1); - const transform = dag.task("transform", async (rows: number, region: string) => `${region}`); - - const extracted = extract(); - transform(extracted, "us"); - - expect(getDagTaskInputs(dag).get("transform")).toEqual({ arg0: extracted, arg1: "us" }); - expect(Object.keys(getDagTaskInputs(dag).get("transform")!)).toEqual(["arg0", "arg1"]); - }); - - it("names positional arguments from argBindings", () => { - const dag = new Dag("example_dag"); - const extract = dag.task("extract", async (): Promise => 1); - const transform = dag.task("transform", async (rows: number, region: string) => `${region}`, { - argBindings: ["rows", "region"], - }); - - const extracted = extract(); - transform(extracted, "us"); - - expect(getDagTaskInputs(dag).get("transform")).toEqual({ rows: extracted, region: "us" }); - }); - - it("labels the arguments argBindings does not reach", () => { + it.each([ + ["a bare value", 7], + ["an array of values", [1, "us"]], + ])("rejects %s where a task's inputs belong", (_label, inputs) => { const dag = new Dag("example_dag"); - const transform = dag.task("transform", async (rows: number, region: string) => `${region}`, { - argBindings: ["rows"], - }); - - transform(1, "us"); + const load = dag.task("load", async ({ total }: { total: number }) => total); - expect(getDagTaskInputs(dag).get("transform")).toEqual({ rows: 1, arg1: "us" }); + expect(() => load(inputs as never)).toThrowError( + /takes one object naming its inputs: myTask\({ rows, region }\)/, + ); }); - it.each([ - [ - "not an array", - { argBindings: 1 }, - /argBindings for Dag "d" task "t" must be an array of names/, - ], - ["not a string", { argBindings: [1] }, /holds 1; each name must be a non-empty string/], - ["empty", { argBindings: [""] }, /holds ""; each name must be a non-empty string/], - ["a number", { argBindings: ["0"] }, /holds "0"; each name must be a non-empty string/], - [ - "a duplicate", - { argBindings: ["a", "a"] }, - /argBindings for Dag "d" task "t" names "a" twice/, - ], - ])("rejects argBindings that are %s", (_label, options, expected) => { - const dag = new Dag("d"); - - expect(() => dag.task("t", async (a: number) => a, options as never)).toThrowError(expected); - }); - - it("reads a single argument that is not a map of names as one positional input", () => { + it("rejects inputs given as more than one argument", () => { const dag = new Dag("example_dag"); - const when = new Date(); - const transform = dag.task("transform", async (at: Date) => at); - - transform(when as never); + const transform = dag.task( + "transform", + async ({ rows, region }: { rows: number; region: string }) => `${region}${rows}`, + ); - expect(getDagTaskInputs(dag).get("transform")).toEqual({ arg0: when }); + // A positional call from plain JavaScript; TypeScript rejects it. + expect(() => (transform as (...inputs: unknown[]) => unknown)({ rows: 1 }, "us")).toThrowError( + /was given 2 arguments; it takes one object naming its inputs/, + ); }); - it("reads a single reference as one positional input rather than a map of names", () => { + it("rejects a bare reference where a task's inputs belong", () => { const dag = new Dag("example_dag"); const extract = dag.task("extract", async (): Promise<{ rows: number }> => ({ rows: 1 })); - const load = dag.task("load", async (totals: { rows: number }) => totals.rows); + const load = dag.task("load", async ({ totals }: { totals: { rows: number } }) => totals.rows); - const extracted = extract(); - load(extracted); - - expect(getDagTaskInputs(dag).get("load")).toEqual({ arg0: extracted }); + expect(() => load(extract() as never)).toThrowError(/takes one object naming its inputs/); }); - it("spreads a positional task's bound arguments back into its argument list", async () => { + it("rejects a handler that declares more than one parameter", () => { const dag = new Dag("example_dag"); - const seen: unknown[] = []; - const transform = dag.task( - "transform", - async (rows: number, region: string) => { - seen.push(rows, region); - }, - { argBindings: ["rows", "region"] }, - ); - - transform(1, "us"); - // The order the runtime hands the bound arguments over in, which is the - // order they were recorded. - await new Bundle(dag).getTaskHandler("example_dag", "transform")!({ - rows: 1, - region: "us", - } as never); - expect(seen).toEqual([1, "us"]); + expect(() => + dag.task("transform", ((rows: number, region: string) => `${region}${rows}`) as never), + ).toThrowError(/declares 2 parameters; a handler takes one object of named arguments/); }); it("calls a task declaring one object of named arguments with that object", async () => { @@ -447,7 +389,6 @@ describe("Dag", () => { it.each([ ["a misspelling", "retry"], ["the raw schema key", "retry_delay"], - ["the positional task_id", "taskId"], ])("rejects %s in the task spec", (_label, key) => { const dag = new Dag("example_dag"); expect(() => @@ -480,6 +421,96 @@ describe("Dag", () => { expect(dag.taskIds).toEqual([]); }); + describe("an omitted task id", () => { + it("takes the handler's function name", () => { + const dag = new Dag("named_dag"); + dag.task(async function extract() {})(); + + expect(dag.taskIds).toEqual(["extract"]); + }); + + it("takes the name of a handler declared elsewhere", () => { + async function transform() {} + const dag = new Dag("named_dag"); + dag.task(transform)(); + + expect(dag.taskIds).toEqual(["transform"]); + }); + + it("still accepts a spec as the second argument", () => { + const dag = new Dag("named_dag"); + dag.task(async function extract() {}, { retries: 2 })(); + + expect(getDagTaskRecords(dag).get("extract")?.spec).toEqual({ retries: 2 }); + }); + + it("is taken from the spec when one names the task", () => { + const dag = new Dag("specced_id_dag"); + dag.task(async function extract() {}, { taskId: "extract_rows" })(); + + expect(dag.taskIds).toEqual(["extract_rows"]); + }); + + it("prefers the positional id over the handler name", () => { + const dag = new Dag("positional_dag"); + dag.task("extract_rows", async function extract() {})(); + + expect(dag.taskIds).toEqual(["extract_rows"]); + }); + + it("rejects a positional id and a spec id together", () => { + // The positional one used to win and the spec's was dropped in silence. + const dag = new Dag("two_ids_dag"); + + expect(() => + dag.task("extract_rows", async function extract() {}, { taskId: "from_spec" }), + ).toThrowError( + /Task "extract_rows" of Dag "two_ids_dag" also carries taskId "from_spec" in its spec/, + ); + expect(dag.taskIds).toEqual([]); + }); + + it("fails for an anonymous handler, naming the two ways to give it an id", () => { + const dag = new Dag("anonymous_dag"); + + expect(() => dag.task(async () => 42)).toThrowError( + /A task of Dag "anonymous_dag" has no id: its handler has no name/, + ); + expect(dag.taskIds).toEqual([]); + }); + + it("points at the bundler when a handler's name was minified away", () => { + const dag = new Dag("minified_dag"); + // What a bundle built by something other than airflow-ts-pack can hold: + // the function is real, but the bundler took its name. + const minified = Object.defineProperty(async () => 42, "name", { value: "" }); + + expect(() => dag.task(minified)).toThrowError( + /A bundler that drops function names also lands here/, + ); + }); + + it("rejects a handler name Airflow would not accept as an id", () => { + const dag = new Dag("bound_dag"); + async function extract({ rows }: { rows: number }) { + return rows; + } + + // `bind` names the result "bound extract", which has a space in it. + expect(() => dag.task(extract.bind(null))).toThrowError( + /would take the id "bound extract" from its handler's name, which Airflow does not accept/, + ); + }); + + it("rejects a non-function where a handler belongs", () => { + const dag = new Dag("bad_handler_dag"); + + expect(() => dag.task("x", 42 as unknown as () => Promise)).toThrowError( + /handler for Dag "bad_handler_dag" task "x" must be a function/, + ); + }); + }); + it("exposes its task IDs in attachment order", () => { const dag = new Dag("ordered_dag"); expect(dag.taskIds).toEqual([]);