kaxil commented on code in PR #71144:
URL: https://github.com/apache/airflow/pull/71144#discussion_r3740704527
##########
ts-sdk/src/sdk/registry.ts:
##########
@@ -17,85 +17,72 @@
* under the License.
*/
+import { Dag, getDagTaskRecords, type Task } from "./dag.js";
import type { TaskHandler } from "./task.js";
-// Mirrors the Python task-SDK KEY_REGEX and validate_key in
airflow.sdk.definitions._internal.node.
-const KEY_REGEX = /^[\p{L}\p{N}_.-]+$/u;
-const MAX_KEY_LENGTH = 250;
-
-function validateKey(name: string, value: string): void {
- if (typeof value !== "string" || !KEY_REGEX.test(value)) {
- throw new Error(
- `${name} must be made of alphanumeric characters, dashes, dots, and
underscores`,
- );
- }
- if (value.length > MAX_KEY_LENGTH) {
- throw new Error(`${name} must be less than ${MAX_KEY_LENGTH} characters,
not ${value.length}`);
- }
-}
-
-/** Identifies the Airflow task handled by a TypeScript function. */
-export interface TaskRegistration {
- /** Identifier of the Dag containing this task. */
+/** A registered Dag with its task IDs, empty Dags included — used for
+ * manifests, where a task-less Dag must stay visible. */
+export interface RegisteredDag {
+ /** Identifier of the registered Dag. */
readonly dagId: string;
- /** Airflow task ID, including any TaskGroup prefix. */
- readonly taskId: string;
+ /** Airflow task IDs, including any TaskGroup prefix. */
+ readonly tasks: string[];
}
-/** Registry of TypeScript task handlers keyed by Dag ID and task ID. */
-export class TaskRegistry {
- readonly #tasks = new Map<string, Map<string, TaskHandler>>();
+/**
+ * Registry of Dag instances keyed by Dag ID.
+ *
+ * Lookups delegate live to each Dag's task map, so tasks added to a Dag
+ * after registration are visible — the registry records Dag identity, not
+ * a snapshot of its tasks.
+ */
+export class DagRegistry {
+ readonly #dags = new Map<string, Dag>();
- /**
- * Register a TypeScript handler for an Airflow task.
- *
- * `dagId` must match the Python Dag's `dag_id`. `taskId` must match the
- * Dag-side operator's `task_id` exactly, including any TaskGroup prefix.
- */
- register<TReturn = unknown>(registration: TaskRegistration, handler:
TaskHandler<TReturn>): void {
- const { dagId, taskId } = registration;
- validateKey("dagId", dagId);
- validateKey("taskId", taskId);
- if (typeof handler !== "function") {
- throw new Error(`handler for Dag "${dagId}" task "${taskId}" must be a
function`);
+ /** Register Dags. Registering an already-registered `dagId` throws,
+ * and a call that throws registers none of its Dags. */
+ register(...dags: Dag[]): void {
+ const incoming = new Set<string>();
+ for (const dag of dags) {
+ if (!(dag instanceof Dag)) {
+ throw new Error("only Dag instances can be registered");
+ }
+ if (this.#dags.has(dag.dagId) || incoming.has(dag.dagId)) {
+ throw new Error(`Dag "${dag.dagId}" is already registered`);
+ }
+ incoming.add(dag.dagId);
}
- const dagTasks = this.#tasks.get(dagId) ?? new Map<string, TaskHandler>();
- if (dagTasks.has(taskId)) {
- throw new Error(`Task "${taskId}" is already registered for Dag
"${dagId}"`);
+ for (const dag of dags) {
+ this.#dags.set(dag.dagId, dag);
}
- dagTasks.set(taskId, handler as TaskHandler);
- this.#tasks.set(dagId, dagTasks);
}
/** Look up a registered handler. Returns `undefined` when no handler
exists. */
- get(dagId: string, taskId: string): TaskHandler | undefined {
- return this.#tasks.get(dagId)?.get(taskId);
+ getTaskHandler(dagId: string, taskId: string): TaskHandler | undefined {
+ const dag = this.#dags.get(dagId);
+ return dag ? getDagTaskRecords(dag).get(taskId)?.handler : undefined;
}
- /** List all registered tasks. */
- list(): TaskRegistration[] {
- return [...this.#tasks.entries()].flatMap(([dagId, tasks]) =>
- [...tasks.keys()].map((taskId) => ({ dagId, taskId })),
+ /** List the task handles across registered Dags. */
+ listTasks(): Task[] {
+ return [...this.#dags.values()].flatMap((dag) =>
+ [...getDagTaskRecords(dag).values()].map((record) => record.task),
);
}
-}
-const defaultRegistry = new TaskRegistry();
-
-/** Register a TypeScript handler in the default task registry. */
-export function registerTask<TReturn = unknown>(
- registration: TaskRegistration,
- handler: TaskHandler<TReturn>,
-): void {
- defaultRegistry.register(registration, handler);
+ /** List every registered Dag with its task IDs, empty Dags included. */
+ listDags(): RegisteredDag[] {
+ return [...this.#dags.values()].map((dag) => ({
+ dagId: dag.dagId,
+ tasks: [...getDagTaskRecords(dag).keys()],
+ }));
+ }
}
-/** Look up a registered handler. Returns `undefined` when no handler exists.
*/
-export function getRegisteredTask(dagId: string, taskId: string): TaskHandler
| undefined {
- return defaultRegistry.get(dagId, taskId);
-}
+/** The registry `registerDags` writes to and the coordinator reads from. */
+export const defaultRegistry = new DagRegistry();
-/** List all registered tasks. */
-export function listRegisteredTasks(): TaskRegistration[] {
- return defaultRegistry.list();
+/** Record Dags in the default registry so the coordinator can run their
tasks. */
+export function registerDags(...dags: Dag[]): void {
Review Comment:
Splitting declaration from registration makes this third step forgettable,
and the chain fails quietly when it is missed.
`buildBundleManifest()` reads `defaultRegistry.listDags()`, so a Dag that
was constructed and populated but never passed here is absent from the
manifest. `runPack` catches only the all-or-nothing case (`registered no
Dags`), so registering 3 of 4 Dags packs successfully.
`NodeCoordinator._find_bundle` ignores the manifest's `dags` map when selecting
a bundle (`task-sdk/src/airflow/sdk/coordinators/node/coordinator.py`), so the
bundle is still launched for the missing Dag's tasks, `handleTask` finds no
handler, and it returns `{type: "TaskState", state: "removed"}`. The user sees
task instances quietly marked *removed*, which normally means the task is no
longer in the Dag, with only a `warning` line in the task log.
Fair caveat: forgetting `import "./sales/tasks"` under the old side-effect
pattern registered nothing either, so this is a new instance of an existing
mistake rather than a regression in kind. What is new is that it is now cheaply
detectable, because the Dag is an object the SDK can see.
Cheapest fix that keeps this shape: have the `Dag` constructor push `this`
onto a module-level list, emit it in `--airflow-metadata` mode, and have pack
fail with `Dag "billing" was declared but never passed to registerDags(...)`.
The structural version is worth a thought too. Java uses this same
constructor shape (`java.rst`: `var dag = new Dag("my_dag");
dag.addTask("fetch", FetchTask.class); return List.of(dag);`) but returns the
Dags from `getDags()`, so omission is unrepresentable; Go inverts ownership
instead, with `Registry.AddDag(dagId) Dag`. A TypeScript analogue of either,
`startCoordinator({ dags: [salesDag, billingDag] })` or an `addDag(dagId,
spec?)` that registers eagerly and returns the handle, removes the forgettable
step rather than detecting it.
On naming, while the package is still unpublished: `Task` here is an opaque
`{dagId, taskId}` identity, whereas Go's `Task` is the executable thing the
runtime calls and Java's is the interface a task class implements. `TaskRef` or
`TaskHandle` would keep the vocabulary aligned across the three SDKs and leave
`Task` free for the real task object native Dag declaration will need.
##########
ts-sdk/src/sdk/dag.ts:
##########
@@ -0,0 +1,208 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// The Dag authoring surface: `new Dag(dagId)` plus `dag.task(taskId,
handler)`.
+
+import type { TaskHandler } from "./task.js";
+
+// Mirrors the Python task-SDK KEY_REGEX and validate_key in
airflow.sdk.definitions._internal.node.
+const KEY_REGEX = /^[\p{L}\p{N}_.-]+$/u;
+const MAX_KEY_LENGTH = 250;
+
+function validateKey(name: string, value: string): void {
+ if (typeof value !== "string" || !KEY_REGEX.test(value)) {
+ throw new Error(
+ `${name} must be made of alphanumeric characters, dashes, dots, and
underscores`,
+ );
+ }
+ if (value.length > MAX_KEY_LENGTH) {
+ throw new Error(`${name} must be less than ${MAX_KEY_LENGTH} characters,
not ${value.length}`);
+ }
+}
+
+/**
+ * Dag-level options.
+ *
+ * Empty today: native TypeScript Dag declaration (schedule, tags, ...) will
add
+ * optional fields here without changing the `Dag` constructor signature.
+ */
+// eslint-disable-next-line @typescript-eslint/no-empty-object-type --
extension point for future native-Dag fields
+export interface DagSpec {}
+
+/**
+ * Task-level options.
+ *
+ * Empty today: future task fields (retries, ...) will land here without
+ * changing the `dag.task()` signature.
+ */
+// eslint-disable-next-line @typescript-eslint/no-empty-object-type --
extension point for future task fields
+export interface TaskSpec {}
+
+/**
+ * Opaque handle to a task registered on a {@link Dag}.
+ *
+ * Pass it as a downstream task's input to declare that the downstream task
+ * consumes this task's return value:
+ *
+ * ```ts
+ * const extracted = dag.task("extract", extractFn);
+ * const transformed = dag.task("transform", transformFn, { inputs: {
extracted } });
+ * dag.task("load", loadFn, { inputs: { transformed } });
+ * ```
+ *
+ * In today's Python-stub mode the stub Dag still defines task order; declared
+ * inputs are retained for the serialized Dag JSON that native TypeScript Dag
+ * declaration will emit. The handler is intentionally not exposed on the
handle.
+ */
+export interface Task {
+ /** Identifier of the Dag this task belongs to. */
+ readonly dagId: string;
+ /** Airflow task ID, including any TaskGroup prefix. */
+ readonly taskId: string;
+}
+
+/**
+ * Upstream task handles a task consumes, keyed by input name.
+ *
+ * Values must be handles returned by `dag.task(...)`. Literal values are
+ * deliberately out of scope for now; the future native-Dag work decides how
+ * they are declared.
+ */
+export type TaskInputs = Readonly<Record<string, Task>>;
+
+/**
+ * Named options for `dag.task()`.
+ *
+ * Keyword-only so neither field has to be positioned around the other, and so
+ * future fields can be added without a new parameter.
+ */
+export interface TaskOptions {
+ /** Upstream task handles whose return values this task consumes. */
+ readonly inputs?: TaskInputs;
+ /** Task-level options. */
+ readonly spec?: TaskSpec;
+}
+
+/** Per-task record a Dag retains: the handle, the handler, its spec, and the
+ * upstream handles feeding it. */
+export interface TaskRecord {
+ readonly task: Task;
+ readonly handler: TaskHandler;
+ readonly spec: TaskSpec;
+ /** Upstream handles keyed by input name; empty when the task has no inputs.
*/
+ readonly inputs: TaskInputs;
+}
+
+// Assigned inside Dag's static block: gives package-internal code read access
+// to the #tasks private field without a public accessor on the Dag class.
+let taskRecordsOf: (dag: Dag) => ReadonlyMap<string, TaskRecord>;
+
+/**
+ * A Dag declared in TypeScript.
+ *
+ * Today the Dag structure itself is still declared by a Python stub file; a
+ * `Dag` instance binds TypeScript handlers to that stub's Dag/task IDs. The
+ * instance retains its `spec` and every task's `(taskId, handler, spec)` so a
+ * future `serialize()` can produce the serialized Dag JSON for native
+ * TypeScript Dag declaration.
+ */
+export class Dag {
+ /** Identifier of this Dag. Must match the Python Dag's `dag_id`. */
+ readonly dagId: string;
+ /** Dag-level options this instance was constructed with. */
+ readonly spec: DagSpec;
+ readonly #tasks = new Map<string, TaskRecord>();
+
+ static {
+ taskRecordsOf = (dag) => dag.#tasks;
+ }
+
+ constructor(dagId: string, spec: DagSpec = {}) {
+ validateKey("dagId", dagId);
+ this.dagId = dagId;
+ this.spec = spec;
+ }
+
+ /**
+ * Register a TypeScript handler for a task of this Dag.
+ *
+ * `taskId` must match the Dag-side operator's `task_id` exactly, including
+ * any TaskGroup prefix. `options.inputs` names the upstream task handles
whose
+ * return values this task consumes. Returns this task's handle.
+ */
+ task<TReturn = unknown>(
+ taskId: string,
+ handler: TaskHandler<TReturn>,
+ options: TaskOptions = {},
+ ): Task {
+ const { inputs = {}, spec = {} } = options;
+ validateKey("taskId", taskId);
+ if (typeof handler !== "function") {
+ throw new Error(`handler for Dag "${this.dagId}" task "${taskId}" must
be a function`);
+ }
+ if (this.#tasks.has(taskId)) {
+ throw new Error(`Task "${taskId}" is already registered for Dag
"${this.dagId}"`);
+ }
+ this.#validateInputs(taskId, inputs);
+ const task: Task = Object.freeze({ dagId: this.dagId, taskId });
+ this.#tasks.set(taskId, {
+ task,
+ handler: handler as TaskHandler,
+ spec,
+ inputs: Object.freeze({ ...inputs }),
Review Comment:
`inputs` is copied and frozen here, but `spec` on the line above and
`this.spec` in the constructor are stored by reference, and
`tests/sdk/dag.test.ts` locks that in with `toBe(dagSpec)` / `toBe(taskSpec)`.
Once `DagSpec` carries real fields, a caller who mutates their spec object
after construction silently mutates the Dag. `Object.freeze({ ...spec })` in
both places, with those two assertions relaxed to `toEqual`, makes the two
options behave alike.
Separately, the test named "records frozen inputs that later mutation of the
caller's object cannot change" does not test the freeze. It adds a new *key* to
the caller's object and asserts the record is unchanged, which passes on the
spread copy alone; delete `Object.freeze` here and it stays green. One
assertion covers it:
`expect(Object.isFrozen(getDagTaskRecords(dag).get("transform")!.inputs)).toBe(true)`.
##########
ts-sdk/src/sdk/dag.ts:
##########
@@ -0,0 +1,208 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// The Dag authoring surface: `new Dag(dagId)` plus `dag.task(taskId,
handler)`.
+
+import type { TaskHandler } from "./task.js";
+
+// Mirrors the Python task-SDK KEY_REGEX and validate_key in
airflow.sdk.definitions._internal.node.
+const KEY_REGEX = /^[\p{L}\p{N}_.-]+$/u;
+const MAX_KEY_LENGTH = 250;
+
+function validateKey(name: string, value: string): void {
+ if (typeof value !== "string" || !KEY_REGEX.test(value)) {
+ throw new Error(
+ `${name} must be made of alphanumeric characters, dashes, dots, and
underscores`,
+ );
+ }
+ if (value.length > MAX_KEY_LENGTH) {
+ throw new Error(`${name} must be less than ${MAX_KEY_LENGTH} characters,
not ${value.length}`);
+ }
+}
+
+/**
+ * Dag-level options.
+ *
+ * Empty today: native TypeScript Dag declaration (schedule, tags, ...) will
add
+ * optional fields here without changing the `Dag` constructor signature.
+ */
+// eslint-disable-next-line @typescript-eslint/no-empty-object-type --
extension point for future native-Dag fields
+export interface DagSpec {}
+
+/**
+ * Task-level options.
+ *
+ * Empty today: future task fields (retries, ...) will land here without
+ * changing the `dag.task()` signature.
+ */
+// eslint-disable-next-line @typescript-eslint/no-empty-object-type --
extension point for future task fields
+export interface TaskSpec {}
+
+/**
+ * Opaque handle to a task registered on a {@link Dag}.
+ *
+ * Pass it as a downstream task's input to declare that the downstream task
+ * consumes this task's return value:
+ *
+ * ```ts
+ * const extracted = dag.task("extract", extractFn);
+ * const transformed = dag.task("transform", transformFn, { inputs: {
extracted } });
+ * dag.task("load", loadFn, { inputs: { transformed } });
+ * ```
+ *
+ * In today's Python-stub mode the stub Dag still defines task order; declared
+ * inputs are retained for the serialized Dag JSON that native TypeScript Dag
+ * declaration will emit. The handler is intentionally not exposed on the
handle.
+ */
+export interface Task {
+ /** Identifier of the Dag this task belongs to. */
+ readonly dagId: string;
+ /** Airflow task ID, including any TaskGroup prefix. */
+ readonly taskId: string;
+}
+
+/**
+ * Upstream task handles a task consumes, keyed by input name.
+ *
+ * Values must be handles returned by `dag.task(...)`. Literal values are
+ * deliberately out of scope for now; the future native-Dag work decides how
+ * they are declared.
+ */
+export type TaskInputs = Readonly<Record<string, Task>>;
+
+/**
+ * Named options for `dag.task()`.
+ *
+ * Keyword-only so neither field has to be positioned around the other, and so
+ * future fields can be added without a new parameter.
+ */
+export interface TaskOptions {
+ /** Upstream task handles whose return values this task consumes. */
+ readonly inputs?: TaskInputs;
+ /** Task-level options. */
+ readonly spec?: TaskSpec;
+}
+
+/** Per-task record a Dag retains: the handle, the handler, its spec, and the
+ * upstream handles feeding it. */
+export interface TaskRecord {
+ readonly task: Task;
+ readonly handler: TaskHandler;
+ readonly spec: TaskSpec;
+ /** Upstream handles keyed by input name; empty when the task has no inputs.
*/
+ readonly inputs: TaskInputs;
+}
+
+// Assigned inside Dag's static block: gives package-internal code read access
+// to the #tasks private field without a public accessor on the Dag class.
+let taskRecordsOf: (dag: Dag) => ReadonlyMap<string, TaskRecord>;
+
+/**
+ * A Dag declared in TypeScript.
+ *
+ * Today the Dag structure itself is still declared by a Python stub file; a
+ * `Dag` instance binds TypeScript handlers to that stub's Dag/task IDs. The
+ * instance retains its `spec` and every task's `(taskId, handler, spec)` so a
+ * future `serialize()` can produce the serialized Dag JSON for native
+ * TypeScript Dag declaration.
+ */
+export class Dag {
+ /** Identifier of this Dag. Must match the Python Dag's `dag_id`. */
+ readonly dagId: string;
+ /** Dag-level options this instance was constructed with. */
+ readonly spec: DagSpec;
+ readonly #tasks = new Map<string, TaskRecord>();
Review Comment:
With tasks in a `#`-private field and `listRegisteredTasks` removed, no
public API is left that answers "which tasks did I wire up?".
`defaultRegistry`, `DagRegistry.listTasks`, `listDags` and `getDagTaskRecords`
are all module-exported but absent from `src/index.ts`, the two-entry `exports`
map blocks deep imports, and `./coordinator` re-exports only
`startCoordinator`, `StartCoordinatorOptions` and `SUPERVISOR_API_VERSION`.
That matters here specifically, because it removes the one test a user could
write against the drift this stub-Dag model is exposed to: that their
TypeScript handlers match the Python `@task.stub` names. It is also the test
that would catch a forgotten `registerDags` at author time instead of as task
instances marked *removed*. The Go SDK exposes this deliberately, through
`EnumerableBundle.OrderedDags()`.
```ts
/** Task IDs attached to this Dag, in attachment order. */
get taskIds(): readonly string[] { return [...this.#tasks.keys()]; }
```
That also lets `registry.listDags()` stop going through `getDagTaskRecords`.
##########
ts-sdk/src/sdk/dag.ts:
##########
@@ -0,0 +1,208 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// The Dag authoring surface: `new Dag(dagId)` plus `dag.task(taskId,
handler)`.
+
+import type { TaskHandler } from "./task.js";
+
+// Mirrors the Python task-SDK KEY_REGEX and validate_key in
airflow.sdk.definitions._internal.node.
+const KEY_REGEX = /^[\p{L}\p{N}_.-]+$/u;
+const MAX_KEY_LENGTH = 250;
+
+function validateKey(name: string, value: string): void {
+ if (typeof value !== "string" || !KEY_REGEX.test(value)) {
+ throw new Error(
+ `${name} must be made of alphanumeric characters, dashes, dots, and
underscores`,
+ );
+ }
+ if (value.length > MAX_KEY_LENGTH) {
+ throw new Error(`${name} must be less than ${MAX_KEY_LENGTH} characters,
not ${value.length}`);
+ }
+}
+
+/**
+ * Dag-level options.
+ *
+ * Empty today: native TypeScript Dag declaration (schedule, tags, ...) will
add
+ * optional fields here without changing the `Dag` constructor signature.
+ */
+// eslint-disable-next-line @typescript-eslint/no-empty-object-type --
extension point for future native-Dag fields
+export interface DagSpec {}
+
+/**
+ * Task-level options.
+ *
+ * Empty today: future task fields (retries, ...) will land here without
+ * changing the `dag.task()` signature.
+ */
+// eslint-disable-next-line @typescript-eslint/no-empty-object-type --
extension point for future task fields
+export interface TaskSpec {}
+
+/**
+ * Opaque handle to a task registered on a {@link Dag}.
+ *
+ * Pass it as a downstream task's input to declare that the downstream task
Review Comment:
`inputs` does not declare anything today, but this headline says it does,
and the gap can produce wrong data rather than an error.
Grepping `ts-sdk/src/`, nothing outside this file reads `TaskRecord.inputs`:
`registry.ts` reads `.handler`, `.task` and `.keys()`, `manifest.ts` reads only
`dagId`/`tasks`. There is also no channel to deliver an input to a handler,
since `TaskHandlerArgs` is still `{ctx, client}` (`src/sdk/task.ts`).
The reachable failure: a user who reads this headline concludes the value is
wired, so writes `await client.getXCom({ key: "return_value" })` with no
`taskId`. Per `src/sdk/client-types.ts:32-33`, `taskId` defaults to the running
task's context, so that reads the task's *own* XCom, returns `null`, and the
task succeeds with the wrong value. No error, no warning, no failed task
instance.
Neither `ts-sdk/README.md` nor `typescript.rst` mentions `inputs`, and the
rst note next to the paragraph this PR edits still says dependencies are
declared in the Python stub Dag, so autocomplete plus this JSDoc is the only
discovery channel.
Either resolution works:
- Drop `inputs`/`TaskInputs`/`#validateInputs` for now and land `Dag` +
`dag.task(taskId, handler)` + `registerDags`, which is the whole refactor the
title promises. Adding an optional trailing `options` parameter later is not
source-breaking in TypeScript, so waiting costs nothing.
- Keep it, but lead the headline with what it does today (nothing), and add
a `dag.task` options table to the rst marking `inputs` reserved and inert.
##########
ts-sdk/src/cli/pack.ts:
##########
@@ -193,9 +202,19 @@ export async function runPack(argv: readonly string[]):
Promise<void> {
});
const manifest = readBundleManifest(stagingPath);
- if (Object.keys(manifest.dags).length === 0) {
+ const dagEntries = Object.entries(manifest.dags);
+ if (dagEntries.length === 0) {
+ throw new Error(
+ `${args.entry} registered no Dags; register Dags with
registerDags(...) before startCoordinator()`,
+ );
+ }
+ const emptyDags = dagEntries
+ .filter(([, dag]) => dag.tasks.length === 0)
+ .map(([dagId]) => dagId);
+ if (emptyDags.length > 0) {
Review Comment:
This is a policy divergence from the packer this file says it mirrors. For
the same condition `airflow-go-pack` prints `warning: dag %q has no tasks` and
continues (`go-sdk/cmd/airflow-go-pack/pack.go`), while this throws. The two
agree on rejecting zero Dags, and the shared schema permits an empty list,
since `dagEntry.tasks` has no `minItems`.
Failing closed is arguably the better call, and there is a forward-looking
argument for it, since `executable/coordinator.py` claims bundle ownership from
`set(dags.keys())` alone and Node is slated to grow the same routing. But it
means one placeholder Dag, or conditional attachment like `if
(process.env.FEATURE_X) dag.task(...)`, blocks the whole bundle build in
TypeScript and not in Go. Worth picking one deliberately and applying it across
both: match Go's warning, or hard-error in both and add `"minItems": 1` to the
schema.
##########
ts-sdk/src/sdk/dag.ts:
##########
@@ -0,0 +1,208 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// The Dag authoring surface: `new Dag(dagId)` plus `dag.task(taskId,
handler)`.
+
+import type { TaskHandler } from "./task.js";
+
+// Mirrors the Python task-SDK KEY_REGEX and validate_key in
airflow.sdk.definitions._internal.node.
+const KEY_REGEX = /^[\p{L}\p{N}_.-]+$/u;
+const MAX_KEY_LENGTH = 250;
+
+function validateKey(name: string, value: string): void {
+ if (typeof value !== "string" || !KEY_REGEX.test(value)) {
+ throw new Error(
+ `${name} must be made of alphanumeric characters, dashes, dots, and
underscores`,
+ );
+ }
+ if (value.length > MAX_KEY_LENGTH) {
+ throw new Error(`${name} must be less than ${MAX_KEY_LENGTH} characters,
not ${value.length}`);
+ }
+}
+
+/**
+ * Dag-level options.
+ *
+ * Empty today: native TypeScript Dag declaration (schedule, tags, ...) will
add
+ * optional fields here without changing the `Dag` constructor signature.
+ */
+// eslint-disable-next-line @typescript-eslint/no-empty-object-type --
extension point for future native-Dag fields
+export interface DagSpec {}
+
+/**
+ * Task-level options.
+ *
+ * Empty today: future task fields (retries, ...) will land here without
+ * changing the `dag.task()` signature.
+ */
+// eslint-disable-next-line @typescript-eslint/no-empty-object-type --
extension point for future task fields
+export interface TaskSpec {}
+
+/**
+ * Opaque handle to a task registered on a {@link Dag}.
+ *
+ * Pass it as a downstream task's input to declare that the downstream task
+ * consumes this task's return value:
+ *
+ * ```ts
+ * const extracted = dag.task("extract", extractFn);
+ * const transformed = dag.task("transform", transformFn, { inputs: {
extracted } });
+ * dag.task("load", loadFn, { inputs: { transformed } });
+ * ```
+ *
+ * In today's Python-stub mode the stub Dag still defines task order; declared
+ * inputs are retained for the serialized Dag JSON that native TypeScript Dag
+ * declaration will emit. The handler is intentionally not exposed on the
handle.
+ */
+export interface Task {
+ /** Identifier of the Dag this task belongs to. */
+ readonly dagId: string;
+ /** Airflow task ID, including any TaskGroup prefix. */
+ readonly taskId: string;
+}
+
+/**
+ * Upstream task handles a task consumes, keyed by input name.
+ *
+ * Values must be handles returned by `dag.task(...)`. Literal values are
+ * deliberately out of scope for now; the future native-Dag work decides how
+ * they are declared.
+ */
+export type TaskInputs = Readonly<Record<string, Task>>;
+
+/**
+ * Named options for `dag.task()`.
+ *
+ * Keyword-only so neither field has to be positioned around the other, and so
+ * future fields can be added without a new parameter.
+ */
+export interface TaskOptions {
+ /** Upstream task handles whose return values this task consumes. */
+ readonly inputs?: TaskInputs;
+ /** Task-level options. */
+ readonly spec?: TaskSpec;
+}
+
+/** Per-task record a Dag retains: the handle, the handler, its spec, and the
+ * upstream handles feeding it. */
+export interface TaskRecord {
+ readonly task: Task;
+ readonly handler: TaskHandler;
+ readonly spec: TaskSpec;
+ /** Upstream handles keyed by input name; empty when the task has no inputs.
*/
+ readonly inputs: TaskInputs;
+}
+
+// Assigned inside Dag's static block: gives package-internal code read access
+// to the #tasks private field without a public accessor on the Dag class.
+let taskRecordsOf: (dag: Dag) => ReadonlyMap<string, TaskRecord>;
+
+/**
+ * A Dag declared in TypeScript.
+ *
+ * Today the Dag structure itself is still declared by a Python stub file; a
+ * `Dag` instance binds TypeScript handlers to that stub's Dag/task IDs. The
+ * instance retains its `spec` and every task's `(taskId, handler, spec)` so a
+ * future `serialize()` can produce the serialized Dag JSON for native
+ * TypeScript Dag declaration.
+ */
+export class Dag {
+ /** Identifier of this Dag. Must match the Python Dag's `dag_id`. */
+ readonly dagId: string;
+ /** Dag-level options this instance was constructed with. */
+ readonly spec: DagSpec;
+ readonly #tasks = new Map<string, TaskRecord>();
+
+ static {
+ taskRecordsOf = (dag) => dag.#tasks;
+ }
+
+ constructor(dagId: string, spec: DagSpec = {}) {
+ validateKey("dagId", dagId);
+ this.dagId = dagId;
+ this.spec = spec;
+ }
+
+ /**
+ * Register a TypeScript handler for a task of this Dag.
+ *
+ * `taskId` must match the Dag-side operator's `task_id` exactly, including
+ * any TaskGroup prefix. `options.inputs` names the upstream task handles
whose
+ * return values this task consumes. Returns this task's handle.
+ */
+ task<TReturn = unknown>(
+ taskId: string,
+ handler: TaskHandler<TReturn>,
+ options: TaskOptions = {},
+ ): Task {
+ const { inputs = {}, spec = {} } = options;
Review Comment:
This destructure accepts any other shape silently, which is out of step with
the three checks around it. `taskId`, `handler`, and every `inputs` value are
runtime-validated precisely because TypeScript can be bypassed, but
`dag.task("transform", fn, { input: { extracted } })` (singular typo) or `{
inpts: {...} }` registers a task with no declared upstream and says nothing.
Reachable from plain JavaScript, since `airflow-ts-pack` bundles whatever
esbuild accepts, and from an `as TaskOptions` cast. This package already fails
closed on the same class of mistake: `parsePackArgs` throws `Unknown option
${arg}`.
Worth fixing in the same place: the new comment at
`tests/public-api.test.ts:178` says "these constructor/method misuses also
throw at runtime", and that is false for one of the five lines it covers.
`dag.task("transform2", async () => undefined, { upstream })` does not throw;
it registers with zero inputs and discards the declared upstream. The other
four do throw. Because the arrow function is never invoked, nothing catches it.
```ts
for (const key of Object.keys(options)) {
if (key !== "inputs" && key !== "spec") {
throw new Error(`Unknown option "${key}" for Dag "${this.dagId}" task
"${taskId}"`);
}
}
```
Also worth handling while here: `options === null` currently fails with an
opaque "Cannot destructure property", because the default only applies to
`undefined`.
##########
ts-sdk/src/cli/pack.ts:
##########
@@ -154,6 +154,15 @@ function readBundleManifest(bundlePath: string):
BundleManifest {
if (!manifest.supervisor_schema_version || !manifest.dags || typeof
manifest.dags !== "object") {
throw new Error(`Bundle produced incomplete ${AIRFLOW_METADATA_FLAG}
output`);
}
+ // The line is whatever the bundle printed, so the per-Dag shape is not
+ // guaranteed by the type assertion above.
+ for (const [dagId, dag] of Object.entries(manifest.dags)) {
+ if (dag == null || !Array.isArray(dag.tasks)) {
Review Comment:
This stops one level short of the schema it protects. Element types are
unchecked, so `{"dags":{"d":{"tasks":[null,{},"",123]}}}` passes here and flows
through `renderMetadataYaml` into the base64 trailer, while
`task-sdk/docs/airflow-metadata.schema.json` defines `dagEntry.tasks.items` as
`{"type": "string", "minLength": 1}`. The Go reference packer cannot emit that,
because it decodes into `Tasks []string`, and the Python reader does not close
the gap either: `_bundle_metadata.py` validates only that the document is a
mapping and that `sdk.supervisor_schema_version` is a non-empty string, so bad
task ids propagate past pack.
```ts
if (dag == null || !Array.isArray(dag.tasks) ||
dag.tasks.some((t) => typeof t !== "string" || t.length === 0)) {
```
Cheap while you are here: the guard on the line above passes for an *array*,
which `Object.entries` then turns into Dags named `"0"`, `"1"`. Pre-existing,
but this loop is the natural place to close it.
##########
ts-sdk/src/sdk/dag.ts:
##########
@@ -0,0 +1,208 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// The Dag authoring surface: `new Dag(dagId)` plus `dag.task(taskId,
handler)`.
+
+import type { TaskHandler } from "./task.js";
+
+// Mirrors the Python task-SDK KEY_REGEX and validate_key in
airflow.sdk.definitions._internal.node.
+const KEY_REGEX = /^[\p{L}\p{N}_.-]+$/u;
+const MAX_KEY_LENGTH = 250;
+
+function validateKey(name: string, value: string): void {
+ if (typeof value !== "string" || !KEY_REGEX.test(value)) {
+ throw new Error(
+ `${name} must be made of alphanumeric characters, dashes, dots, and
underscores`,
+ );
+ }
+ if (value.length > MAX_KEY_LENGTH) {
+ throw new Error(`${name} must be less than ${MAX_KEY_LENGTH} characters,
not ${value.length}`);
+ }
+}
+
+/**
+ * Dag-level options.
+ *
+ * Empty today: native TypeScript Dag declaration (schedule, tags, ...) will
add
+ * optional fields here without changing the `Dag` constructor signature.
+ */
+// eslint-disable-next-line @typescript-eslint/no-empty-object-type --
extension point for future native-Dag fields
+export interface DagSpec {}
Review Comment:
`interface DagSpec {}` is the empty object type, so `spec` accepts every
non-nullish value, and the rule suppressed here is the one that reports exactly
that. Today `new Dag("d", 42)` and `dag.task("t", fn, { spec: { retries: 3 }
})` both typecheck and store the value unvalidated. `TaskSpec`'s own docstring
advertises `retries` as a future field, so that second call is a natural thing
to write and have silently ignored.
The repo has already taken a position on the spelling:
`airflow-core/src/airflow/ui/rules/typescript.js:668` sets
`@typescript-eslint/no-empty-object-type` to `ERROR`, and its doc block lists
`type FooType = {}` as incorrect while prescribing `type FooType = object`.
That is the UI package's config rather than `ts-sdk/eslint.config.js`, so
precedent rather than a rule this PR breaks, but the question has an in-repo
answer.
The stated rationale also inverts. Once `DagSpec` gains its first field,
necessarily optional, it becomes a weak type, and any call site passing an
object with no overlapping property starts failing with TS2559. Every other
extension the comment anticipates is already non-breaking: an optional trailing
parameter, an optional property on `TaskOptions` (a type users pass and never
implement), and `Task` becoming `Task<TReturn = unknown>` are all
source-compatible. The only source-breaking change in that set is the one the
placeholder introduces.
`export type DagSpec = object` rejects primitives and matches the spelling
the repo prescribes; `Record<string, never>` also rejects `{retries: 3}`.
Either one removes both `eslint-disable` lines.
One caveat on my side: I verified the lint policy and these declarations,
but not the assignability behaviour with a `tsc` run, since `ts-sdk/` has no
installed `node_modules` in my checkout. Both are one-line checks.
--
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]
