shivaam commented on code in PR #74484: URL: https://github.com/apache/airflow/pull/74484#discussion_r4235796149
########## ts-sdk/README.md: ########## @@ -175,6 +175,94 @@ bundle the coordinator's `task_handler_bundle_name` names. See the guide for all See [`example/`](https://github.com/apache/airflow/tree/main/ts-sdk/example) for a working project that serves both a Dag declared in TypeScript and the tasks of two Python Dags from one bundle. +## Human-in-the-loop tasks + +A Dag declared in TypeScript can wait for a person, as `HITLOperator` and `ApprovalOperator` do in +Python. `approval(...)` and `hitl(...)` come from the `apache-airflow-ts-sdk/hitl` subpath and are +passed to `dag.task` in place of a handler. Each is named after its Python operator without +"Operator", and each option is the Python keyword argument in camelCase: + +```ts +import { Bundle, Dag } from "apache-airflow-ts-sdk"; +import { approval } from "apache-airflow-ts-sdk/hitl"; + +const dag = new Dag("release", { queue: "typescript" }); +const report = dag.task("build_report", async () => ({ version: "1.4" }))(); + +const signOff = dag.task( + "sign_off", + approval({ + subject: ({ report }: { report: { version: string } }) => `Ship ${report.version}?`, + responseTimeout: 4 * 60 * 60, // seconds + defaults: "Reject", // the answer given when the timeout passes + }), +)({ report }); +signOff.before(dag.task("publish", async () => {})()); + +await new Bundle(dag).serve(); +``` + +The task's first run writes the request and parks the task in `awaiting_input`; its Node process +exits, so no worker slot is held while it waits. Once someone responds on the Required Actions page, +or the timeout passes, the task resumes in a new process. It does not write the request again or run +`subject` or `body` again. + +- `subject` and `body` (Markdown) are a string, or a function of the task's inputs, run once. They + may read different inputs, and the task takes both. +- `responseTimeout` is in whole **seconds**. With `defaults`, the defaults are applied when it passes; + without, the task fails, and its retries apply. A HITL task does not enforce `executionTimeout`, so + `dag.task` rejects it and asks for `responseTimeout` instead. +- `assignedUsers: [{ id, name }]` limits who can respond, by the auth manager's user id. +- The task returns the response, which downstream tasks receive and XCom stores under + `return_value`: `{ chosen_options, params_input, responded_at, responded_by_user, timedout }`. + These are the keys of Python's `HITLOperator` result plus `timedout`; `responded_at` is an + ISO-8601 string, and `responded_by_user` is `null` when the timeout defaults were applied. + +**On "Reject", an approval succeeds and skips the tasks directly downstream by default**, as +`ApprovalOperator` does. The run then shows as successful, with the guarded tasks skipped. Set +`failOnReject: true` to fail the approval instead, or `ignoreDownstreamTriggerRules: true` to skip +every task downstream. `hitl(...)` offers any `options`, with `multiple` for several, and never +skips or fails on what is chosen, even an option named "Reject". + +A failed resume retries, and the retry writes a fresh request, so `approval({ failOnReject: true })` +with `retries` asks the reviewer again on each attempt. + +When the task resumes, it reads its options, its `params` and its reject options from the Dag it +runs from, as `HITLOperator` does. A versioned Dag bundle resumes it from the same version it +started on; with an unversioned bundle, a bundle deployed while the request waits is what the +response is read against. Clearing the task starts a new attempt, which writes a fresh request. + +### Forms + +`params` adds form fields to the request, as `HITLOperator`'s `params` does. Each field is a `value` +that pre-fills it, an optional `description`, and an optional JSON Schema that the form renders: + +```ts +import { hitl } from "apache-airflow-ts-sdk/hitl"; + +const rollout = dag.task( + "size_rollout", + hitl({ + subject: "Approve the rollout?", + options: ["Approve", "Reject"], + defaults: ["Reject"], + params: { + replicas: { + value: 3, + description: "How many replicas to start", + schema: { type: "integer", minimum: 1, maximum: 10 }, + }, + }, + }), +)(); +``` + +The answers arrive as `params_input`, keyed by the param names. When the timeout applies `defaults`, Review Comment: Fixed -- 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]
