Class: Ask::Graph

Inherits:
Object
  • Object
show all
Defined in:
lib/ask/graph.rb,
lib/ask/graph/runner.rb,
lib/ask/graph/context.rb,
lib/ask/graph/version.rb

Overview

Define durable workflow graphs with named steps, conditional routing, parallel execution, sub-graph composition, human-in-the-loop approval, per-item checkpointing loops, step timeouts, retry policies, and lifecycle hooks.

Examples:

Basic workflow

class HandleCall::Workflow < Ask::Graph
  step Transcribe
  step Classify, if: :needs_classification?
  step BookAppointment
end

Sub-graph composition

# Define a reusable workflow
module NotifyCustomer
  class Workflow < Ask::Graph
    step SendEmail
    step LogNotification
  end
end

# Wrap it in a PORO step
module OrderFulfillment
  class NotifyCustomer
    def call(context)
      NotifyCustomer::Workflow.call(context)
    end
  end
end

# Compose — every step is a PORO
class OrderFulfillment::Workflow < Ask::Graph
  step ValidatePayment
  step NotifyCustomer
  step ShipOrder
end

With timeout and retry

class ApiWorkflow < Ask::Graph
  step FetchData, timeout: 30, retry: 3
  step ProcessData, description: "Transform raw data into reports"
end

With lifecycle hooks

class MonitoredWorkflow < Ask::Graph
  before_step :log_start
  after_step  :log_completion
  on_failure  :alert_team

  step ValidateOrder
  step ProcessPayment, timeout: 15, retry: 2
end

Defined Under Namespace

Classes: Context, Paused, Runner, StepFailed, StepTimeout, WorkflowTimeout

Constant Summary collapse

VERSION = 
"0.7.1"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(input = nil, storage: nil) ⇒ Graph

Returns a new instance of Graph.



284
285
286
287
288
289
290
291
292
293
294
295
 
# File 'lib/ask/graph.rb', line 284

def initialize(input = nil, storage: nil)
  @input = input
  store = storage || self.class.storage || Ask::State::Memory.new
  hooks = self.class.lifecycle_hooks
  @runner = Runner.new(self.class.declarations,
                       storage: store,
                       hooks: hooks,
                       graph_instance: self,
                       step_timeout: self.class.step_timeout,
                       workflow_timeout: self.class.workflow_timeout)
  @context = nil
end

Instance Attribute Details

#context ⇒ Object (readonly)

Returns the value of attribute context.



298
299
300
 
# File 'lib/ask/graph.rb', line 298

def context
  @context
end

#runner ⇒ Object (readonly)

Returns the value of attribute runner.



297
298
299
 
# File 'lib/ask/graph.rb', line 297

def runner
  @runner
end

Class Method Details

.after_step(method_name) ⇒ Object

Register a callback to run after every successful step.

Parameters:

  • method_name (Symbol) —

    method name on the graph class



192
193
194
 
# File 'lib/ask/graph.rb', line 192

def after_step(method_name)
  lifecycle_hooks[:after_step] << method_name
end

.approve(klass, **opts) ⇒ Object

Declare a human-in-the-loop step. Runs the step, then pauses and waits for external input via #resume.

Parameters:

  • klass (Class) —

    a class that responds to #call(context)

  • opts (Hash) —

    optional if:, unless:, description:, timeout:, retry:



238
239
240
 
# File 'lib/ask/graph.rb', line 238

def approve(klass, **opts)
  declarations << build_declaration(:approve, klass, opts)
end

.before_step(method_name) ⇒ Object

Register a callback to run before every step.

Parameters:

  • method_name (Symbol) —

    method name on the graph class



186
187
188
 
# File 'lib/ask/graph.rb', line 186

def before_step(method_name)
  lifecycle_hooks[:before_step] << method_name
end

.call(input = nil, storage: nil) ⇒ Object

Create an instance and run it.

When input is a Context, the graph is run as a sub-graph: the context data is exported, the graph runs with it, and results are merged back into the original context automatically.

Examples:

Standalone — pass data

OrderFulfillment::Workflow.call({ user: "alice" })

Inside a step — pass the outer context

class ValidatePayment
  def call(context)
    PaymentGateway::Workflow.call(context)
  end
end


257
258
259
260
261
262
263
264
265
 
# File 'lib/ask/graph.rb', line 257

def call(input = nil, storage: nil)
  if input.is_a?(Context)
    ctx = new(input.export_data, storage: storage).call
    input.import(ctx)
    ctx
  else
    new(input, storage: storage).call
  end
end

.declarations ⇒ Object

All step declarations collected across the class hierarchy.



64
65
66
 
# File 'lib/ask/graph.rb', line 64

def declarations
  @declarations ||= []
end

.inherited(subclass) ⇒ Object

Inherited hook — copies declarations to subclasses.



69
70
71
72
73
74
 
# File 'lib/ask/graph.rb', line 69

def inherited(subclass)
  super
  subclass.instance_variable_set(:@declarations, declarations.dup)
  subclass.instance_variable_set(:@lifecycle_hooks,
    lifecycle_hooks.transform_values(&:dup))
end

.lifecycle_hooks ⇒ Object

--- Lifecycle hooks ---



180
181
182
 
# File 'lib/ask/graph.rb', line 180

def lifecycle_hooks
  @lifecycle_hooks ||= { before_step: [], after_step: [], on_failure: [] }
end

.on_failure(method_name) ⇒ Object

Register a callback to run when a step fails.

Parameters:

  • method_name (Symbol) —

    method name on the graph class



198
199
200
 
# File 'lib/ask/graph.rb', line 198

def on_failure(method_name)
  lifecycle_hooks[:on_failure] << method_name
end

.run ⇒ Object

Create an instance and run it.

When input is a Context, the graph is run as a sub-graph: the context data is exported, the graph runs with it, and results are merged back into the original context automatically.

Examples:

Standalone — pass data

OrderFulfillment::Workflow.call({ user: "alice" })

Inside a step — pass the outer context

class ValidatePayment
  def call(context)
    PaymentGateway::Workflow.call(context)
  end
end


266
267
268
269
270
271
272
273
274
 
# File 'lib/ask/graph.rb', line 266

def call(input = nil, storage: nil)
  if input.is_a?(Context)
    ctx = new(input.export_data, storage: storage).call
    input.import(ctx)
    ctx
  else
    new(input, storage: storage).call
  end
end

.step(klass, **opts) ⇒ Object

Declare a sequential step.

Parameters:

  • klass (Class) —

    a class that responds to #call(context)

  • conditions (Hash) —

    optional if: or unless: condition method name, description: human-readable label, timeout: seconds, retry: number of retries on failure



210
211
212
 
# File 'lib/ask/graph.rb', line 210

def step(klass, **opts)
  declarations << build_declaration(:step, klass, opts)
end

.step_timeout(seconds = :not_set) ⇒ Integer? Also known as: default_step_timeout

Set or get the default timeout (in seconds) for each step in this graph. Steps without an explicit timeout: option inherit this value. Setting on Ask::Graph itself applies to all graphs that don't set their own default — a global default.

Examples:

Global default for all graphs

Ask::Graph.default_step_timeout 30

Per-graph override

class MyGraph < Ask::Graph
  step_timeout 15
  step FastOp          # uses 15s
  step SlowOp, timeout: 60  # overrides
end

Parameters:

  • seconds (Integer, nil) (defaults to: :not_set) —

    timeout in seconds, or nil to clear

Returns:

  • (Integer, nil) —

    the current default step timeout for this graph



95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/ask/graph.rb', line 95

def step_timeout(seconds = :not_set)
  if seconds == :not_set
    if instance_variable_defined?(:@step_timeout)
      @step_timeout
    elsif superclass.respond_to?(:step_timeout)
      superclass.step_timeout
    end
  else
    @step_timeout = seconds
  end
end

.steps(*klasses, **opts) ⇒ Object

Declare parallel steps — all run simultaneously.

Parameters:

  • klasses (Array<Class>) —

    step classes to run in parallel

  • opts (Hash) —

    optional if:, unless:, timeout: seconds, retry: number of retries



219
220
221
222
223
224
225
226
227
228
229
230
 
# File 'lib/ask/graph.rb', line 219

def steps(*klasses, **opts)
  declarations << {
    classes: klasses,
    type: :steps,
    name: klasses.map { |k| k.name || k.to_s }.join(", "),
    if: opts[:if],
    unless: opts[:unless],
    description: opts[:description],
    timeout: opts[:timeout],
    retry: opts[:retry]
  }
end

.storage(store = :not_set) ⇒ Object?

Set or get the default storage backend for checkpoint data. The storage object must respond to #set(key, value) and #get(key).

Setting on Ask::Graph itself applies to all graphs that don't set their own storage — a global default.

Examples:

Global default for all graphs

Ask::Graph.storage RedisPool.new

Per-graph override

class MyGraph < Ask::Graph
  storage PostgresStore.new
end

Per-call override

MyGraph.call(input, storage: InMemory.new)

Parameters:

  • store (Object, nil) (defaults to: :not_set) —

    a backend that responds to #set and #get

Returns:

  • (Object, nil) —

    the current storage for this graph



160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/ask/graph.rb', line 160

def storage(store = :not_set)
  if store == :not_set
    if instance_variable_defined?(:@storage)
      @storage
    elsif superclass.respond_to?(:storage)
      superclass.storage
    end
  else
    @storage = store
  end
end

.storage=(store) ⇒ Object

Assignment form of #storage — convenient in initializers:

Ask::Graph.storage = RedisPool.new


174
175
176
 
# File 'lib/ask/graph.rb', line 174

def storage=(store)
  @storage = store
end

.workflow_timeout(seconds = :not_set) ⇒ Integer? Also known as: default_workflow_timeout

Set or get the total runtime cap (in seconds) for the entire graph. The whole workflow aborts with WorkflowTimeout if it exceeds this limit, regardless of individual step timeouts. Setting on Ask::Graph itself applies to all graphs that don't set their own default — a global default.

Examples:

Global default for all graphs

Ask::Graph.default_workflow_timeout 60

Per-graph override

class MyGraph < Ask::Graph
  workflow_timeout 30
  step FetchData, timeout: 10
  step ProcessData, timeout: 10
end

Parameters:

  • seconds (Integer, nil) (defaults to: :not_set) —

    timeout in seconds, or nil to clear

Returns:

  • (Integer, nil) —

    the current workflow timeout for this graph



126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/ask/graph.rb', line 126

def workflow_timeout(seconds = :not_set)
  if seconds == :not_set
    if instance_variable_defined?(:@workflow_timeout)
      @workflow_timeout
    elsif superclass.respond_to?(:workflow_timeout)
      superclass.workflow_timeout
    end
  else
    @workflow_timeout = seconds
  end
end

Instance Method Details

#call ⇒ Object Also known as: run 



300
301
302
303
304
305
306
307
 
# File 'lib/ask/graph.rb', line 300

def call(*)
  @context = Context.new(self, @input)
  @runner.run(@context)
  @context
rescue => e
  raise unless e.is_a?(Paused)
  @context
end

#resume(input:) ⇒ Object 



310
311
312
313
314
 
# File 'lib/ask/graph.rb', line 310

def resume(input:)
  @context.resume_input = input.to_s
  @runner.run(@context)
  @context
end