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

Constant Summary collapse

VERSION = 
"0.6.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.



246
247
248
249
250
251
252
253
254
255
256
 
# File 'lib/ask/graph.rb', line 246

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,
                       default_timeout: self.class.timeout)
  @context = nil
end

Instance Attribute Details

#context ⇒ Object (readonly)

Returns the value of attribute context.



259
260
261
 
# File 'lib/ask/graph.rb', line 259

def context
  @context
end

#runner ⇒ Object (readonly)

Returns the value of attribute runner.



258
259
260
 
# File 'lib/ask/graph.rb', line 258

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



154
155
156
 
# File 'lib/ask/graph.rb', line 154

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:



200
201
202
 
# File 'lib/ask/graph.rb', line 200

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



148
149
150
 
# File 'lib/ask/graph.rb', line 148

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


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

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



142
143
144
 
# File 'lib/ask/graph.rb', line 142

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



160
161
162
 
# File 'lib/ask/graph.rb', line 160

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


228
229
230
231
232
233
234
235
236
 
# File 'lib/ask/graph.rb', line 228

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



172
173
174
 
# File 'lib/ask/graph.rb', line 172

def step(klass, **opts)
  declarations << build_declaration(:step, klass, opts)
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



181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/ask/graph.rb', line 181

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



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

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

.timeout(seconds = :not_set) ⇒ Integer?

Set or get the default timeout (in seconds) for steps in this graph. Steps without an explicit timeout: option will 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.timeout 30

Per-graph override

class MyGraph < Ask::Graph
  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 timeout for this graph



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

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

Instance Method Details

#call ⇒ Object Also known as: run 



261
262
263
264
265
266
267
268
 
# File 'lib/ask/graph.rb', line 261

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

#resume(input:) ⇒ Object 



271
272
273
274
275
 
# File 'lib/ask/graph.rb', line 271

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