Class: Hatchet::Workflow

Inherits:
Object
  • Object
show all
Defined in:
lib/hatchet/workflow.rb,
sig/hatchet/workflow.rbs

Overview

Represents a workflow definition with one or more tasks arranged in a DAG.

Examples:

Define a simple workflow

wf = hatchet.workflow(name: "MyWorkflow")
step1 = wf.task(:step1) { |input, ctx| { "value" => 42 } }
wf.task(:step2, parents: [step1]) { |input, ctx|
  { "result" => ctx.task_output(step1)["value"] + 1 }
}

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name:, on_events: [], on_crons: [], concurrency: nil, default_priority: nil, task_defaults: nil, default_filters: [], sticky: nil, idempotency: nil, client: nil) ⇒ Workflow

Returns a new instance of Workflow.

Parameters:

  • name (String) —

    Workflow name

  • on_events (Array<String>) (defaults to: []) —

    Event trigger keys

  • on_crons (Array<String>) (defaults to: []) —

    Cron trigger expressions

  • concurrency (Array<ConcurrencyExpression>, ConcurrencyExpression, nil) (defaults to: nil)
  • default_priority (Integer, nil) (defaults to: nil) —

    Default priority

  • task_defaults (Hash, nil) (defaults to: nil) —

    Default task settings

  • default_filters (Array<DefaultFilter>) (defaults to: []) —

    Default filters

  • sticky (Symbol, nil) (defaults to: nil) —

    Sticky strategy

  • idempotency (Hatchet::TTLBasedIdempotencyConfig, Hatchet::StatusBasedIdempotencyConfig, nil) (defaults to: nil) —

    Idempotency configuration

  • client (Hatchet::Client, nil) (defaults to: nil) —

    The client



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/hatchet/workflow.rb', line 65

def initialize(
  name:,
  on_events: [],
  on_crons: [],
  concurrency: nil,
  default_priority: nil,
  task_defaults: nil,
  default_filters: [],
  sticky: nil,
  idempotency: nil,
  client: nil
)
  @name = name
  @tasks = {}
  @on_events = on_events
  @on_crons = on_crons
  @concurrency = concurrency
  @default_priority = default_priority
  @task_defaults = task_defaults
  @default_filters = default_filters
  @sticky = sticky
  @idempotency = idempotency
  @client = client
  @on_failure = nil
  @on_success = nil
  @id = nil
end

Instance Attribute Details

#client ⇒ Hatchet::Client? (readonly)

Returns The Hatchet client.

Returns:



41
42
43
 
# File 'lib/hatchet/workflow.rb', line 41

def client
  @client
end

#concurrency ⇒ Array<ConcurrencyExpression>, ... (readonly)

Returns Workflow-level concurrency.

Returns:



26
27
28
 
# File 'lib/hatchet/workflow.rb', line 26

def concurrency
  @concurrency
end

#default_filters ⇒ Array<DefaultFilter> (readonly)

Returns Default filters for event triggers.

Returns:



35
36
37
 
# File 'lib/hatchet/workflow.rb', line 35

def default_filters
  @default_filters
end

#default_priority ⇒ Integer? (readonly)

Returns Default priority for runs (1-4).

Returns:

  • (Integer, nil) —

    Default priority for runs (1-4)



29
30
31
 
# File 'lib/hatchet/workflow.rb', line 29

def default_priority
  @default_priority
end

#id ⇒ String?

Get the workflow ID (UUID). If not already set, lazily resolves it by looking up the workflow by name via the REST API.

Returns:

  • (String, nil) —

    The workflow UUID



97
98
99
 
# File 'lib/hatchet/workflow.rb', line 97

def id
  @id ||= resolve_workflow_id
end

#idempotency ⇒ Hatchet::TTLBasedIdempotencyConfig, ... (readonly)

Returns Idempotency configuration.

Returns:



50
51
52
 
# File 'lib/hatchet/workflow.rb', line 50

def idempotency
  @idempotency
end

#name ⇒ String (readonly)

Returns Workflow name.

Returns:

  • (String) —

    Workflow name



14
15
16
 
# File 'lib/hatchet/workflow.rb', line 14

def name
  @name
end

#on_crons ⇒ Array<String> (readonly)

Returns Cron expressions that trigger this workflow.

Returns:

  • (Array<String>) —

    Cron expressions that trigger this workflow



23
24
25
 
# File 'lib/hatchet/workflow.rb', line 23

def on_crons
  @on_crons
end

#on_events ⇒ Array<String> (readonly)

Returns Event keys that trigger this workflow.

Returns:

  • (Array<String>) —

    Event keys that trigger this workflow



20
21
22
 
# File 'lib/hatchet/workflow.rb', line 20

def on_events
  @on_events
end

#on_failure ⇒ Task? (readonly)

Returns The on_failure task.

Returns:

  • (Task, nil) —

    The on_failure task



44
45
46
 
# File 'lib/hatchet/workflow.rb', line 44

def on_failure
  @on_failure
end

#on_success ⇒ Task? (readonly)

Returns The on_success task.

Returns:

  • (Task, nil) —

    The on_success task



47
48
49
 
# File 'lib/hatchet/workflow.rb', line 47

def on_success
  @on_success
end

#sticky ⇒ Symbol? (readonly)

Returns Sticky strategy (:soft, :hard).

Returns:

  • (Symbol, nil) —

    Sticky strategy (:soft, :hard)



38
39
40
 
# File 'lib/hatchet/workflow.rb', line 38

def sticky
  @sticky
end

#task_defaults ⇒ Hash? (readonly)

Returns Default task settings.

Returns:

  • (Hash, nil) —

    Default task settings



32
33
34
 
# File 'lib/hatchet/workflow.rb', line 32

def task_defaults
  @task_defaults
end

#tasks ⇒ Hash<Symbol, Task> (readonly)

Returns Map of task name to Task object.

Returns:

  • (Hash<Symbol, Task>) —

    Map of task name to Task object



17
18
19
 
# File 'lib/hatchet/workflow.rb', line 17

def tasks
  @tasks
end

Instance Method Details

#batch_task(name, batch:, **opts) {|inputs, ctx| ... } ⇒ Task

Define a batch task within this workflow.

Batch tasks buffer concurrent runs until Hatchet flushes the batch (size reached or flush interval), then invoke the block once with all buffered inputs keyed by each run's task-run external id. The block must return a Hash mapping each id to its output, or use broadcast_output on the batch config to return the same result to all callers. retries is always forced to 0 for batch tasks.

Preview: batch tasks are in beta and may change in future releases.

Parameters:

  • name (Symbol, String) —

    Task name

  • batch (Hatchet::BatchTaskConfig) —

    Batch configuration

  • opts (Hash) —

    Other Task options forwarded to #task.

Yields:

  • (inputs, ctx) —

    The batch execution block, receiving a Hash of task-run external id => input

Returns:

  • (Task) —

    The created batch task



161
162
163
 
# File 'lib/hatchet/workflow.rb', line 161

def batch_task(name, batch:, **opts, &)
  task(name, batch: batch, **opts, &)
end

#create_bulk_run_item(input: {}, key: nil, options: nil) ⇒ Hash

Create a bulk run item for this workflow, intended to be used with the #run_many methods.

Parameters:

  • input (Hash) (defaults to: {}) —

    The input data for the workflow

  • key (String, nil) (defaults to: nil) —

    The key for the workflow run, used for identification and deduplication

  • options (TriggerWorkflowOptions, nil) (defaults to: nil) —

    Additional options for the workflow run

  • input: (Hash[String, untyped]) (defaults to: {})
  • key: (String, nil) (defaults to: nil)
  • options: (TriggerWorkflowOptions, nil) (defaults to: nil)

Returns:

  • (Hash) —

    A bulk run item that can be passed to the run_many methods



313
314
315
316
317
318
 
# File 'lib/hatchet/workflow.rb', line 313

def create_bulk_run_item(input: {}, key: nil, options: nil)
  item = { input: input }
  item[:key] = key if key
  item[:options] = options if options
  item
end

#create_cron(cron_name, expression, input: {}) ⇒ Object

Create a cron trigger for this workflow.

Parameters:

  • cron_name (String) —

    The name of the cron job

  • expression (String) —

    The cron expression that defines the schedule

  • input (Hash) (defaults to: {}) —

    The input data for the workflow

  • input: (Hash[String, untyped]) (defaults to: {})

Returns:

  • (Object) —

    The created cron workflow trigger

Raises:



340
341
342
343
344
345
346
347
348
349
 
# File 'lib/hatchet/workflow.rb', line 340

def create_cron(cron_name, expression, input: {})
  raise Error, "No client associated with workflow #{@name}" unless @client

  @client.cron.create(
    workflow_name: @name,
    cron_name: cron_name,
    expression: expression,
    input: input,
  )
end

#durable_task(name, eviction_policy: Hatchet::DEFAULT_DURABLE_TASK_EVICTION_POLICY, **opts) {|input, ctx| ... } ⇒ Task

Define a durable task within this workflow.

Parameters:

  • name (Symbol, String) —

    Task name

  • eviction_policy (Hatchet::EvictionPolicy, nil) (defaults to: Hatchet::DEFAULT_DURABLE_TASK_EVICTION_POLICY) —

    Eviction policy for this durable task. Defaults to DEFAULT_DURABLE_TASK_EVICTION_POLICY (15-minute TTL, capacity-eviction enabled). Pass nil to disable eviction entirely for this task.

  • opts (Hash) —

    Other Task options forwarded to #task.

  • eviction_policy: (EvictionPolicy, nil) (defaults to: Hatchet::DEFAULT_DURABLE_TASK_EVICTION_POLICY)

Yields:

  • (input, ctx) —

    The task execution block

Yield Parameters:

Yield Returns:

  • (Object)

Returns:

  • (Task) —

    The created durable task



142
143
144
 
# File 'lib/hatchet/workflow.rb', line 142

def durable_task(name, eviction_policy: Hatchet::DEFAULT_DURABLE_TASK_EVICTION_POLICY, **opts, &)
  task(name, durable: true, eviction_policy: eviction_policy, **opts, &)
end

#on_failure_task(**opts) {|input, ctx| ... } ⇒ Task

Define an on_failure task for this workflow

Parameters:

  • opts (Hash) —

    Task options

Yields:

  • (input, ctx) —

    The on_failure task block

Yield Parameters:

  • arg0 (Hash[String, untyped])
  • arg1 (Context)

Yield Returns:

  • (Object)

Returns:



170
171
172
173
174
175
176
177
178
 
# File 'lib/hatchet/workflow.rb', line 170

def on_failure_task(**opts, &)
  @on_failure = Task.new(
    name: :on_failure,
    workflow: self,
    client: @client,
    **opts,
    &
  )
end

#on_success_task(**opts) {|input, ctx| ... } ⇒ Task

Define an on_success task for this workflow

Parameters:

  • opts (Hash) —

    Task options

Yields:

  • (input, ctx) —

    The on_success task block

Yield Parameters:

  • arg0 (Hash[String, untyped])
  • arg1 (Context)

Yield Returns:

  • (Object)

Returns:



185
186
187
188
189
190
191
192
193
 
# File 'lib/hatchet/workflow.rb', line 185

def on_success_task(**opts, &)
  @on_success = Task.new(
    name: :on_success,
    workflow: self,
    client: @client,
    **opts,
    &
  )
end

#run(input = {}, options: nil) ⇒ Hash

Run this workflow synchronously and wait for it to complete.

Parameters:

  • input (Hash) (defaults to: {}) —

    The input data for the workflow

  • options (TriggerWorkflowOptions, nil) (defaults to: nil) —

    Additional options for workflow execution, such as additional_metadata: and priority:

  • options: (TriggerWorkflowOptions, nil) (defaults to: nil)

Returns:

  • (Hash) —

    The workflow run output, keyed by task name (e.g. {"step1" => {...}, "step2" => {...}})

Raises:



261
262
263
264
265
 
# File 'lib/hatchet/workflow.rb', line 261

def run(input = {}, options: nil)
  raise Error, "No client associated with workflow #{@name}" unless @client

  @client.admin.trigger_workflow(self, input, options: options)
end

#run_many(items, return_exceptions: false) ⇒ Array

Run this workflow in bulk and wait for all runs to complete. Runs are triggered via bulk gRPC triggering (batched by 1000) and results are collected concurrently.

Parameters:

  • items (Array<Hash>) —

    A list of bulk run items, as created by #create_bulk_run_item

  • return_exceptions (Boolean) (defaults to: false) —

    If true, exceptions are returned as part of the results instead of being raised

  • return_exceptions: (Boolean) (defaults to: false)

Returns:

  • (Array) —

    A list of results for each workflow run

Raises:



289
290
291
292
293
 
# File 'lib/hatchet/workflow.rb', line 289

def run_many(items, return_exceptions: false)
  raise Error, "No client associated with workflow #{@name}" unless @client

  @client.admin.trigger_workflow_many(self, items, return_exceptions: return_exceptions)
end

#run_many_no_wait(items) ⇒ Array<WorkflowRunRef>

Run this workflow in bulk without waiting for the runs to complete.

Parameters:

Returns:

  • (Array<WorkflowRunRef>) —

    A list of references to the triggered workflow runs

Raises:



300
301
302
303
304
 
# File 'lib/hatchet/workflow.rb', line 300

def run_many_no_wait(items)
  raise Error, "No client associated with workflow #{@name}" unless @client

  @client.admin.trigger_workflow_many_no_wait(self, items)
end

#run_no_wait(input = {}, options: nil) ⇒ WorkflowRunRef

Trigger a workflow run without waiting for it to complete. Useful for starting a run and immediately returning a reference to it without blocking while the workflow runs.

Parameters:

  • input (Hash) (defaults to: {}) —

    The input data for the workflow

  • options (TriggerWorkflowOptions, nil) (defaults to: nil) —

    Additional options for workflow execution

  • options: (TriggerWorkflowOptions, nil) (defaults to: nil)

Returns:

  • (WorkflowRunRef) —

    A reference to the workflow run, whose result method blocks until the run completes

Raises:



275
276
277
278
279
 
# File 'lib/hatchet/workflow.rb', line 275

def run_no_wait(input = {}, options: nil)
  raise Error, "No client associated with workflow #{@name}" unless @client

  @client.admin.trigger_workflow_no_wait(self, input, options: options)
end

#schedule(time, input: {}, options: nil) ⇒ Object

Schedule this workflow to run at a specific time.

Parameters:

  • time (Time) —

    When to execute the workflow

  • input (Hash) (defaults to: {}) —

    The input data for the workflow

  • options (ScheduleTriggerWorkflowOptions, nil) (defaults to: nil) —

    Additional schedule options

  • input: (Hash[String, untyped]) (defaults to: {})
  • options: (TriggerWorkflowOptions, nil) (defaults to: nil)

Returns:

  • (Object) —

    The schedule response from the Hatchet engine

Raises:



327
328
329
330
331
 
# File 'lib/hatchet/workflow.rb', line 327

def schedule(time, input: {}, options: nil)
  raise Error, "No client associated with workflow #{@name}" unless @client

  @client.admin.schedule_workflow(self, time, input: input, options: options)
end

#task(name, **opts) {|input, ctx| ... } ⇒ Task

Define a task within this workflow. The block receives the workflow input and a Context object, and its return value (a Hash) becomes the task output.

Parameters:

  • name (Symbol, String) —

    The name of the task

  • opts (Hash) —

    a customizable set of options

Options Hash (**opts):

  • :parents (Array<Task, Symbol>) — default: [] —

    A list of tasks that are parents of the task. Note: parents must be defined before their children

  • :execution_timeout (Integer, String, nil) — default: nil —

    The maximum time to wait for the task to complete, in seconds or as a duration string (e.g. "60s")

  • :schedule_timeout (Integer, String, nil) — default: nil —

    The maximum time to wait for the task to be scheduled

  • :retries (Integer, nil) — default: nil —

    The number of times to retry the task before failing

  • :backoff_factor (Float, nil) — default: nil —

    The backoff factor for controlling exponential backoff in retries

  • :backoff_max_seconds (Integer, nil) — default: nil —

    The maximum number of seconds to allow retries with exponential backoff to continue

  • :rate_limits (Array<RateLimit>) — default: [] —

    A list of rate limit configurations for the task

  • :concurrency (ConcurrencyExpression, Array<ConcurrencyExpression>, nil) — default: nil —

    A concurrency expression (or list of them) controlling the concurrency settings for this task

  • :desired_worker_labels (Hash, nil) — default: nil —

    A hash of desired worker labels that determine to which worker the task should be assigned

  • :wait_for (Array) — default: [] —

    A list of conditions that must be met before the task can run

  • :skip_if (Array) — default: [] —

    A list of conditions that, if met, will cause the task to be skipped

  • :deps (Hash, nil) — default: nil —

    Dependency providers to inject into the task's context

Yields:

  • (input, ctx) —

    The task execution block

Yield Parameters:

  • arg0 (Hash[String, untyped])
  • arg1 (Context)

Yield Returns:

  • (Object)

Returns:

  • (Task) —

    The created task



120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/hatchet/workflow.rb', line 120

def task(name, **opts, &)
  t = Task.new(
    name: name,
    workflow: self,
    client: @client,
    **opts,
    &
  )
  @tasks[t.name] = t
  t
end

#to_proto(config) ⇒ V1::CreateWorkflowVersionRequest

Convert this workflow to a V1::CreateWorkflowVersionRequest protobuf message.

Parameters:

Returns:



199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
 
# File 'lib/hatchet/workflow.rb', line 199

def to_proto(config)
  service_name = config.apply_namespace(@name.downcase)

  # Namespace event triggers
  event_triggers = @on_events.map { |e| config.apply_namespace(e) }

  # Convert tasks to proto
  task_protos = @tasks.values.map { |t| t.to_proto(service_name, config: config) }

  # On-failure task
  on_failure_proto = @on_failure&.to_proto(service_name, config: config)

  # Build concurrency
  concurrency_proto = nil
  concurrency_arr = []

  if @concurrency
    conc_list = @concurrency.is_a?(Array) ? @concurrency : [@concurrency]

    if conc_list.length == 1
      concurrency_proto = conc_list.first.to_proto
    else
      concurrency_arr = conc_list.map(&:to_proto)
    end
  end

  # Sticky strategy
  sticky_proto = nil
  if @sticky
    sticky_map = { soft: :SOFT, hard: :HARD }
    sticky_proto = sticky_map[@sticky]
  end

  # Default filters
  filter_protos = (@default_filters || []).map(&:to_proto)

  args = {
    name: config.apply_namespace(@name),
    event_triggers: event_triggers,
    cron_triggers: @on_crons || [],
    tasks: task_protos,
  }

  args[:concurrency] = concurrency_proto if concurrency_proto
  args[:concurrency_arr] = concurrency_arr unless concurrency_arr.empty?
  args[:on_failure_task] = on_failure_proto if on_failure_proto
  args[:sticky] = sticky_proto if sticky_proto
  args[:default_priority] = @default_priority if @default_priority
  args[:default_filters] = filter_protos unless filter_protos.empty?

  args[:idempotency] = @idempotency.to_proto if @idempotency

  ::V1::CreateWorkflowVersionRequest.new(**args)
end