jason810496 commented on code in PR #72047: URL: https://github.com/apache/airflow/pull/72047#discussion_r3955179080
########## ts-sdk/adr/0001-mixed-lang-dag-interface.md: ########## @@ -0,0 +1,89 @@ +<!-- + 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. + --> + +# ADR-0001: Mixed-Lang Dag — TypeScript Task Interface + +## Status + +Proposed + +## Context + +The Python `@task.stub` call site already defines task data flow. TypeScript tasks should consume those bindings directly as named arguments, instead of re-fetching each value with `client.getXCom(...)`. + +This ADR covers only the TypeScript call-site interface. The argument-binding spec itself (its shape, how it's materialized, how it travels over the wire) is a separate, protocol-level decision recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). Given that spec, this ADR only answers what TypeScript code a user writes. + +A mixed-language Dag declares its structure in Python and supplies task bodies from TypeScript, using `MixedLangDag`, a class dedicated to this mode. The native case, where TypeScript owns the graph too, is a separate `Dag` class, covered in [ADR-0002](0002-native-dag-interface.md). + +## Decision + +TypeScript uses one syntax for the TaskFlow binding: named arguments merged onto the handler's single parameter object, alongside the SDK's own `ctx`/`client`. + +```ts +const dag = new MixedLangDag("etl"); + +interface TransformArgs { + region_code: string; + threshold: number; +} + +async function transform({ ctx, client, region_code, threshold }: TransformArgs & TaskHandlerArgs) { + const rows = await client.getXCom<number>({ key: "return_value", taskId: "extract" }); + if (rows === null) { + throw new Error(`task ${ctx.taskId} has no upstream row count to transform`); + } + const passed = rows >= threshold; + await client.setXCom({ key: "region", value: region_code }); + return { region_code, passed }; +} + +dag.task("transform", transform); +``` + +Renaming a Python name that isn't idiomatic TypeScript is ordinary destructuring, not a separate mechanism: + +```ts +async function report({ run_label: runLabel }: { run_label: string } & TaskHandlerArgs) { + if (runLabel !== "nightly") { + throw new Error(`expected run label "nightly" but got "${runLabel}"`); + } +} + Review Comment: I made the arguments transformed automatically by default but users themself are still able to explicitly bind them via `withArgNames`. ########## ts-sdk/adr/0001-mixed-lang-dag-interface.md: ########## @@ -0,0 +1,89 @@ +<!-- + 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. + --> + +# ADR-0001: Mixed-Lang Dag — TypeScript Task Interface + +## Status + +Proposed + +## Context + +The Python `@task.stub` call site already defines task data flow. TypeScript tasks should consume those bindings directly as named arguments, instead of re-fetching each value with `client.getXCom(...)`. + +This ADR covers only the TypeScript call-site interface. The argument-binding spec itself (its shape, how it's materialized, how it travels over the wire) is a separate, protocol-level decision recorded in [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). Given that spec, this ADR only answers what TypeScript code a user writes. + +A mixed-language Dag declares its structure in Python and supplies task bodies from TypeScript, using `MixedLangDag`, a class dedicated to this mode. The native case, where TypeScript owns the graph too, is a separate `Dag` class, covered in [ADR-0002](0002-native-dag-interface.md). + +## Decision + +TypeScript uses one syntax for the TaskFlow binding: named arguments merged onto the handler's single parameter object, alongside the SDK's own `ctx`/`client`. + +```ts +const dag = new MixedLangDag("etl"); + +interface TransformArgs { + region_code: string; + threshold: number; +} + +async function transform({ ctx, client, region_code, threshold }: TransformArgs & TaskHandlerArgs) { + const rows = await client.getXCom<number>({ key: "return_value", taskId: "extract" }); + if (rows === null) { + throw new Error(`task ${ctx.taskId} has no upstream row count to transform`); + } + const passed = rows >= threshold; + await client.setXCom({ key: "region", value: region_code }); + return { region_code, passed }; +} + +dag.task("transform", transform); +``` + +Renaming a Python name that isn't idiomatic TypeScript is ordinary destructuring, not a separate mechanism: + +```ts +async function report({ run_label: runLabel }: { run_label: string } & TaskHandlerArgs) { + if (runLabel !== "nightly") { + throw new Error(`expected run label "nightly" but got "${runLabel}"`); + } +} + +dag.task("report", report); +``` + +### How + +- `MixedLangDag.task()` returns a plain `TaskRef`, not a callable factory. There is nothing to wire, since the Python file already owns task order. Calling it the way a native `Dag`'s factory is called (`transform()`) is a compile error, since `TaskRef` has no call signature, not a runtime throw. +- Wire names match the Python parameter names character for character, with no case- or separator-insensitive fallback. Renaming happens once, at the destructuring site. +- `ctx` and `client` are reserved, permanently: bound arguments are merged flat into the same object alongside them, and a bound name that collides with either fails the task at dispatch. +- An upstream's return value is not delivered as a bound argument. Read it explicitly via `client.getXCom({ key: "return_value", taskId: "..." })`. +- `tsc` cannot check a handler's destructuring pattern against the Python call site. A typo binds `undefined` silently; the runtime logs the bound names at dispatch and includes them in a handler-failure message, so the mismatch is diagnosable from the task log. + +## Open Questions + +- Should the decoded bindings also be exposed as a public, positional/raw accessor (name/value pairs, no interface required), or should that stay an internal runtime detail? +- Should `ctx` and `client` become explicit getter functions (`getClient()`, `getContext()`) instead of arguments merged into the handler's object? (See [ADR-0002](0002-native-dag-interface.md) for where this same question resurfaces on the native-Dag side.) Review Comment: No more `TaskHandlerArgs`, we only allow TaskFlow arguments being injected for now. ########## ts-sdk/adr/0002-native-dag-interface.md: ########## @@ -0,0 +1,143 @@ +<!-- + 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. + --> + +# ADR-0002: Native TypeScript Dag — Interface Design + +## Status + +Proposed + +## Context + +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so TypeScript itself must express both the graph and the task bodies. This ADR covers only what that TypeScript call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case is the separate `MixedLangDag` class, covered in [ADR-0001](0001-mixed-lang-dag-interface.md). This ADR shares the injectable `ctx`/`client` question raised there, and shares its protocol substrate (the argument-binding spec) with [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). + +## Decision + +`dag.task(taskId, handler)` returns a factory. Calling the factory both places the task in the Dag and supplies its arguments by name, the same shape Python TaskFlow itself uses for `load(transformed=transform(...))`: + +```ts +const dag = new Dag("ts_etl"); + +const extract = dag.task("extract", async ({ client }): Promise<number> => { + const rows = 42; + await client.setXCom({ key: "row_count", value: rows }); + return rows; +}); + +interface TransformArgs { + extracted: number; +} + +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); + +interface LoadArgs { + transformed: number; +} + +const load = dag.task("load", async ({ ctx, client, transformed }: LoadArgs & TaskHandlerArgs) => { + if (transformed <= 0) { + throw new Error(`task ${ctx.taskId} received a non-positive value: ${transformed}`); + } + await client.setXCom({ key: "loaded", value: transformed }); +}); + +load({ transformed: transform({ extracted: extract() }) }); +``` + +The call graph is the task graph. `tsc` checks every wired key against the handler's own parameter type, and a `TaskRef` only exists once its producing call has returned, so a cycle is unrepresentable rather than merely rejected by a validator. Every task must be called exactly once; an uncalled task fails when the Dag is read, so a task can't be silently left out of the graph. + +## If `ctx`/`client` were real getter methods instead of injected arguments + +The intersection type above, `TransformArgs & TaskHandlerArgs`, exists for one reason: today's handler signature carries `ctx`/`client` as arguments, so a handler that wants type safety on its own data has to say so explicitly. If `ctx`/`client` came from getter functions instead, a handler's parameter type would be exactly its own data: + +```ts +// Before: ctx/client share the handler's one argument object, so +// TransformArgs must be intersected with TaskHandlerArgs. +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); +``` + +```ts +// After: ctx/client come from getters, so the handler's parameter type is +// exactly its own data, with no TaskHandlerArgs intersection needed. +import { getClient } from "@apache-airflow/ts-sdk"; + +const transform = dag.task("transform", async ({ extracted }: TransformArgs) => { + const client = getClient(); + await client.setXCom({ key: "doubled", value: extracted * 2 }); + return extracted * 2; +}); +``` Review Comment: I agreed, let's go with the following shape. ```ts const extracted = extract(); const transformed = transform({ extracted }); const loaded = load({ transformed }); ``` ########## ts-sdk/adr/0002-native-dag-interface.md: ########## @@ -0,0 +1,143 @@ +<!-- + 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. + --> + +# ADR-0002: Native TypeScript Dag — Interface Design + +## Status + +Proposed + +## Context + +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so TypeScript itself must express both the graph and the task bodies. This ADR covers only what that TypeScript call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case is the separate `MixedLangDag` class, covered in [ADR-0001](0001-mixed-lang-dag-interface.md). This ADR shares the injectable `ctx`/`client` question raised there, and shares its protocol substrate (the argument-binding spec) with [`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). + +## Decision + +`dag.task(taskId, handler)` returns a factory. Calling the factory both places the task in the Dag and supplies its arguments by name, the same shape Python TaskFlow itself uses for `load(transformed=transform(...))`: + +```ts +const dag = new Dag("ts_etl"); + +const extract = dag.task("extract", async ({ client }): Promise<number> => { + const rows = 42; + await client.setXCom({ key: "row_count", value: rows }); + return rows; +}); + +interface TransformArgs { + extracted: number; +} + +const transform = dag.task( + "transform", + async ({ extracted }: TransformArgs & TaskHandlerArgs) => extracted * 2, +); + +interface LoadArgs { + transformed: number; +} + +const load = dag.task("load", async ({ ctx, client, transformed }: LoadArgs & TaskHandlerArgs) => { + if (transformed <= 0) { + throw new Error(`task ${ctx.taskId} received a non-positive value: ${transformed}`); + } + await client.setXCom({ key: "loaded", value: transformed }); +}); + +load({ transformed: transform({ extracted: extract() }) }); Review Comment: We can achieve the `<<, >>` with `.before` and `.after` call. -- 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]
