kaxil commented on code in PR #75358:
URL: https://github.com/apache/airflow/pull/75358#discussion_r4352873589
##########
providers/standard/src/airflow/providers/standard/operators/hitl.py:
##########
@@ -50,11 +66,165 @@
from airflow.sdk.exceptions import TaskAwaitingInput
if TYPE_CHECKING:
+ from datetime import datetime
+
from airflow.providers.common.compat.sdk import Context
from airflow.sdk.execution_time.hitl import HITLUser
from airflow.sdk.types import RuntimeTaskInstanceProtocol
+def _request_input_and_wait(
+ task: BaseOperator,
+ context: Context,
+ *,
+ options: list[str],
+ subject: str,
+ body: str | None,
+ defaults: list[str] | None,
+ multiple: bool,
+ params: dict[str, Any],
+ assigned_users: list[HITLUser] | None,
+ notifiers: Sequence[BaseNotifier],
+ timeout: timedelta | None,
+ timeout_datetime: datetime | None,
+ method_name: str,
+ kwargs: dict[str, Any] | None = None,
+) -> NoReturn:
+ """Write the Human-in-the-loop request, notify, and stop the task until a
response arrives."""
+ ti_id = context["task_instance"].id
+ upsert_hitl_detail(
+ ti_id=ti_id,
+ options=options,
+ subject=subject,
+ body=body,
+ defaults=defaults,
+ multiple=multiple,
+ params=params,
+ assigned_users=assigned_users,
+ )
+
+ task.log.info("Waiting for response")
+ for notifier in notifiers:
+ notifier(context)
+
+ if AIRFLOW_V_3_3_PLUS:
+ # Airflow 3.3+: park the task in AWAITING_INPUT -- no trigger, no
triggerer. The task
+ # is resumed by the Core API response handler or the scheduler timeout
sweep, so the
+ # triggerer no longer needs to run for Human-in-the-loop tasks to make
progress.
+ raise TaskAwaitingInput(method_name=method_name, kwargs=kwargs,
timeout=timeout)
+
+ # Fallback for Airflow versions < 3.3: defer the response check to
HITLTrigger on the triggerer.
+ task.defer(
+ trigger=HITLTrigger(
+ ti_id=ti_id,
+ options=options,
+ defaults=defaults,
+ params=params,
+ multiple=multiple,
+ timeout_datetime=timeout_datetime,
+ ),
+ method_name=method_name,
+ kwargs=kwargs,
+ )
+
+
+def request_input(
+ *,
+ subject: str,
+ options: list[str],
+ body: str | None = None,
+ defaults: str | list[str] | None = None,
+ multiple: bool = False,
+ params: ParamsDict | dict[str, Any] | None = None,
+ notifiers: Sequence[BaseNotifier] | BaseNotifier | None = None,
+ assigned_users: HITLUser | list[HITLUser] | None = None,
+ timeout: timedelta | None = None,
+ key: str | None = None,
+) -> InputResponse:
+ """
+ Ask a person for input from inside a running Python task, and return their
answer.
+
+ The first time a question is asked, the request appears on the Required
Actions page, the same
+ as one from :class:`HITLOperator`, and the task stops there: it waits in
the ``awaiting_input``
+ state without holding a worker slot. Once the request is answered, the
task runs again from the
+ start of its callable, and this call returns the answer instead of asking
again. Save anything
+ the second run should not redo (for example an agent framework's run
state) to
+ ``task_state_store`` before calling this.
+
+ It works in ``@task`` and ``PythonOperator`` callables (sync or async),
including ``@task.branch``
+ and ``@task.short_circuit``. It can't pause tasks that run their callable
in another process, such
+ as ``@task.virtualenv``, ``@task.external_python``, ``@task.docker`` or
``@task.kubernetes``.
+
+ Needs Airflow 3.4+. Earlier versions keep the first request's subject,
body and options when a
+ task asks again, after a resume or on a retry, so the reviewer would
answer a question they were
+ not shown.
+
+ :param subject: Headline shown to the reviewer.
+ :param options: Options the reviewer chooses from.
+ :param body: Details the reviewer needs to decide (Markdown).
+ :param defaults: Options applied if ``timeout`` passes without an answer.
Without defaults, the
+ task fails with
:class:`~airflow.providers.standard.exceptions.HITLTimeoutError` at the timeout.
+ :param multiple: Whether the reviewer can choose more than one option.
+ :param params: Form fields for the reviewer to fill in, defined like Dag
params.
+ :param notifiers: Notifiers called with the task context once the request
is written.
+ :param assigned_users: Users allowed to answer. Anyone with access can
answer if not set.
+ :param timeout: How long to wait for an answer. Waits indefinitely if not
set.
+ :param key: Which question this is, so a resumed task gets each answer
back at the right call.
+ Defaults to ``subject``. Each key is asked at most once per try. Set
it when the subject can
+ change between runs, or when the task can ask the same subject more
than once, for example
+ one approval per agent tool call (use the call's id).
+ """
+ if not AIRFLOW_V_3_4_PLUS:
+ raise AirflowOptionalProviderFeatureException("request_input() needs
Airflow 3.4+.")
+ context = get_current_context()
+ task = context["task"]
+ if not isinstance(task, PythonOperator) or isinstance(task,
_BasePythonVirtualenvOperator):
+ raise TypeError(
+ f"request_input() was called from {type(task).__name__}, which
cannot resume it. It works "
+ "only in tasks that run their Python callable in the task's own
process, such as @task "
+ "and PythonOperator."
+ )
+ defaults = [defaults] if isinstance(defaults, str) else defaults
+ params = params if isinstance(params, ParamsDict) else ParamsDict(params
or {})
+ validate_options(options)
+ validate_params(params)
+ validate_defaults(defaults, options, multiple)
+ key = subject if key is None else key
+ answer_log = answer_log_var.get()
+ if answer_log is not None and (answer := answer_log.take(key)) is not None:
+ # The code may have changed since the reviewer answered; never hand
back an option it no
+ # longer offers.
+ validate_chosen_options(answer["chosen_options"], options)
+ return answer
+ serialized_params = {name: params.get_param(name).serialize() for name in
params}
Review Comment:
Good catch, confirmed it: the parked copy of a param with no default carried
`NOTSET`, which can't be read back on resume. Fixed in 5c4285cbe4d: the parked
copy now carries `None` there (the resume only checks the answer against each
schema, and every key still has to be present), plus a test that parks and
resumes with a required param.
--
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]
