shivaam commented on code in PR #74484: URL: https://github.com/apache/airflow/pull/74484#discussion_r4235821367
########## ts-sdk/src/hitl/operator.ts: ########## @@ -0,0 +1,279 @@ +/*! + * 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. + */ + +// Mirrors `HITLOperator.execute` / `execute_complete` and `ApprovalOperator.execute_complete` +// in the standard provider, on the Airflow 3.3+ path that parks the task in `awaiting_input`. +// +// The task runs twice, in two processes. The first run writes the request and parks; the +// second starts with `next_method` set and the response in `next_kwargs.event`. +// +// As `HITLOperator` does, the second run reads the options, the params and the reject policy from +// the task it is started for, and works out what a rejection skips from that task's Dag; nothing +// is carried over from the first run. A versioned Dag bundle starts both runs from the same +// version. With an unversioned bundle, a bundle deployed while the task waits is the one the +// response is read against, as it is for any task that runs after a deployment. + +import type { JsonValue } from "../sdk/client-types.js"; +import { isPlainRecord, type Dag } from "../sdk/dag.js"; +import { getDagDownstreamTaskIds } from "../sdk/dag-graph.js"; +import { brandOperator, type OperatorContext, type OperatorOutcome } from "../sdk/operator.js"; +import { runInTaskScope } from "../sdk/task.js"; +import type { HITLResult, HITLTask, HITLTaskFields, HITLText, HITLUser } from "./spec.js"; + +export const APPROVE = "Approve"; +export const REJECT = "Reject"; + +/** Build the operator for a task whose options `spec.ts` has checked. */ +export function createHITLTask<TArgs extends object | void>( + fields: HITLTaskFields, +): HITLTask<TArgs> { + const isApproval = fields.kind === "approval"; + const task: HITLTask<never> = { + ...fields, + operatorName: isApproval ? "ApprovalOperator" : "HITLOperator", + // An approval skips what follows it on "Reject". + canSkipDownstream: isApproval, + requiresTaskId: true, + label: "human-in-the-loop", + taskIdExample: isApproval + ? 'dag.task("sign_off", approval({ subject: "..." }))' + : 'dag.task("choose", hitl({ subject: "...", options: ["..."] }))', + executionTimeoutAlternative: "responseTimeout", + execute: (op) => executeHITL(task, op), + executeComplete: (op, event) => resumeHITL(task, op, event), + }; + return brandOperator(task) as HITLTask<TArgs>; +} + +/** The first run: write the request, then park the task until a response arrives. */ +async function executeHITL(task: HITLTask<never>, op: OperatorContext): Promise<OperatorOutcome> { + const { details, ctx, client, logs } = op; + // The inputs are read here only, for the text: the resumed run must not depend on upstream + // XComs still being there. + const args = await op.resolveArgs(task.argNames); + const render = (text: HITLText<never, string | null>) => + runInTaskScope({ ctx, client }, () => renderText(text, args)); + const subject = await render(task.subject); + const body = task.body === undefined ? null : await render(task.body); + if (typeof subject !== "string" || subject.length === 0) { + return op.fail(`The subject of task "${ctx.taskId}" has to be a non-empty string`); + } + if (body !== null && body !== undefined && typeof body !== "string") { Review Comment: Added in c3ecdedf59: a body that returns a number, a non-timeout resume error, an assigned user with extra keys, and a non-object spec. Each test fails with its guard disabled. -- 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]
