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, 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

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

Create a bulk run item for use with run_many

Parameters:

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

    Input data

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

    Deduplication key

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

    Trigger options

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

Returns:

  • (Hash) —

    Bulk run item



271
272
273
274
275
276
 
# File 'lib/hatchet/workflow.rb', line 271

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) —

    Name for the cron

  • expression (String) —

    Cron expression

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

    Workflow input

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

Returns:

  • (Object) —

    Cron result

Raises:



296
297
298
299
300
301
302
303
304
305
 
# File 'lib/hatchet/workflow.rb', line 296

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



129
130
131
 
# File 'lib/hatchet/workflow.rb', line 129

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:



138
139
140
141
142
143
144
145
146
 
# File 'lib/hatchet/workflow.rb', line 138

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:



153
154
155
156
157
158
159
160
161
 
# File 'lib/hatchet/workflow.rb', line 153

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

Parameters:

Returns:

  • (Hash) —

    The workflow run output

Raises:



227
228
229
230
231
 
# File 'lib/hatchet/workflow.rb', line 227

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 many instances of this workflow in bulk

Parameters:

  • items (Array<Hash>) —

    Bulk run items

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

    Return exceptions instead of raising

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

Returns:

  • (Array) —

    Results

Raises:



249
250
251
252
253
 
# File 'lib/hatchet/workflow.rb', line 249

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 many instances without waiting for results

Parameters:

  • items (Array<Hash>) —

    Bulk run items

Returns:

Raises:



259
260
261
262
263
 
# File 'lib/hatchet/workflow.rb', line 259

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

Run this workflow without waiting for the result

Parameters:

Returns:

Raises:



238
239
240
241
242
 
# File 'lib/hatchet/workflow.rb', line 238

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 for future execution

Parameters:

  • time (Time) —

    When to execute

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

    Workflow input

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

    Schedule options

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

Returns:

  • (Object) —

    Schedule result

Raises:



284
285
286
287
288
 
# File 'lib/hatchet/workflow.rb', line 284

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

Parameters:

  • name (Symbol, String) —

    Task name

  • opts (Hash) —

    Task options (parents:, execution_timeout:, retries:, etc.)

Yields:

  • (input, ctx) —

    The task execution block

Yield Parameters:

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

Yield Returns:

  • (Object)

Returns:

  • (Task) —

    The created task



107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/hatchet/workflow.rb', line 107

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:



167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
 
# File 'lib/hatchet/workflow.rb', line 167

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