Skip to main content

interactive_handler

Interactive-DAG JOB_REQUEST handler.

Orchestrates the gRPC worker_only interactive-DAG trigger: detecting the request, building the initiator mailbox, sending the pre-task lifecycle messages (JOB_ACCEPT → PREPARING_DATA → CONFIGURING_TASK), opening the cache Run row, assembling the run context, and scheduling the DAG as a fire-and-forget task.

This lives in the flows.dag package rather than federated.pod so the message-flow intricacies stay off the Pod, which only routes a JOB_REQUEST to _InteractiveDAGHandler and otherwise stays agnostic to how the interactive DAG is driven. It holds a reference to the owning Pod for the shared task-request helpers and pod state (hub, identifier, datasources, mailbox); Pod is imported only under TYPE_CHECKING to avoid an import cycle (pod already imports this package).

Classes​

InteractiveTrigger​

class InteractiveTrigger(**data: Any):

The three things an interactive-DAG run needs from its trigger.

Both trigger paths — a modeller's JOB_REQUEST and the Hub's JOB_REQUEST_FROM_HUB — narrow to this, so _InteractiveDAGHandler.handle is shared between them and stays ignorant of how the trigger was authenticated or where its flow spec came from.

Every field is optional because handle is what reports a missing one back to the initiator (as a TASK_ABORT); rejecting at construction would move that failure before the reply channel exists.

Attributes

  • flow_spec_raw: The already-template-rendered v9 FlowSpec.
  • datasource_name: Datasource to run against, which also selects the background cache partition the interactive phase reads from.
  • project_id: The project this run belongs to, if known.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static datasource_name : str | None
  • static flow_spec_raw : dict[str, typing.Any] | None
  • static model_config
  • static project_id : str | None

Static methods​

from_task_request​

def from_task_request(    task_request_message: _TaskRequestMessage,) ‑> InteractiveTrigger:

Narrow a modeller JOB_REQUEST's task request to a trigger.