Class: Fractor::Workflow::Builder

Inherits:
Object
  • Object
show all
Defined in:
lib/fractor/workflow/builder.rb

Overview

Programmatic API for building workflows without DSL Useful for generating workflows dynamically

Example:

builder = Fractor::Workflow::Builder.new("my-workflow")
builder.input_type(InputData)
builder.output_type(OutputData)
builder.add_job("process", ProcessWorker, inputs: :workflow)
builder.add_job("finalize", FinalizeWorker, needs: "process", inputs: "process")
workflow_class = builder.build
workflow = workflow_class.new
result = workflow.execute(input: data)

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name) ⇒ Builder

Returns a new instance of Builder.



20
21
22
23
24
25
 
# File 'lib/fractor/workflow/builder.rb', line 20

def initialize(name)
  @name = name
  @jobs = []
  @input_type_class = nil
  @output_type_class = nil
end

Instance Attribute Details

#input_type_class ⇒ Object (readonly)

Returns the value of attribute input_type_class.



18
19
20
 
# File 'lib/fractor/workflow/builder.rb', line 18

def input_type_class
  @input_type_class
end

#jobs ⇒ Object (readonly)

Returns the value of attribute jobs.



18
19
20
 
# File 'lib/fractor/workflow/builder.rb', line 18

def jobs
  @jobs
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



18
19
20
 
# File 'lib/fractor/workflow/builder.rb', line 18

def name
  @name
end

#output_type_class ⇒ Object (readonly)

Returns the value of attribute output_type_class.



18
19
20
 
# File 'lib/fractor/workflow/builder.rb', line 18

def output_type_class
  @output_type_class
end

Instance Method Details

#add_job(id, worker, needs: nil, inputs: nil, condition: nil, outputs_to_workflow: false, terminates: false) ⇒ Object

Add a job to the workflow

Parameters:

  • id (String) —

    Job identifier

  • worker (Class) —

    Worker class

  • needs (String, Array<String>) (defaults to: nil) —

    Job dependencies

  • inputs (Symbol, String, Hash) (defaults to: nil) —

    Input configuration

  • condition (Proc) (defaults to: nil) —

    Conditional execution lambda

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

    Whether job outputs to workflow

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

    Whether job terminates workflow



48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/fractor/workflow/builder.rb', line 48

def add_job(id, worker, needs: nil, inputs: nil, condition: nil,
            outputs_to_workflow: false, terminates: false)
  @jobs << {
    id: id,
    worker: worker,
    needs: needs,
    inputs: inputs,
    condition: condition,
    outputs_to_workflow: outputs_to_workflow,
    terminates: terminates,
  }
  self
end

#build ⇒ Object

Build the workflow class



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/fractor/workflow/builder.rb', line 78

def build
  builder_name = @name
  builder_input_type = @input_type_class
  builder_output_type = @output_type_class
  builder_jobs = @jobs.dup

  # Define helper methods that will be available
  find_start_jobs_proc = lambda do |jobs|
    jobs.select { |j| j[:needs].nil? || j[:needs].empty? }
      .map { |j| j[:id] }
  end

  find_end_jobs_proc = lambda do |jobs|
    jobs.select { |j| j[:outputs_to_workflow] || j[:terminates] }
      .map { |j| j[:id] }
  end

  configure_inputs_proc = lambda do |job_dsl, inputs_config|
    return unless inputs_config

    case inputs_config
    when :workflow, "workflow"
      job_dsl.inputs_from_workflow
    when String
      job_dsl.inputs_from_job(inputs_config)
    when Hash
      if inputs_config[:from_job]
        job_dsl.inputs_from_job(inputs_config[:from_job])
      elsif inputs_config[:from_multiple]
        job_dsl.inputs_from_multiple(inputs_config[:from_multiple])
      end
    end
  end

  Class.new(Fractor::Workflow) do
    workflow builder_name do
      input_type builder_input_type if builder_input_type
      output_type builder_output_type if builder_output_type

      # Determine start and end jobs
      start_jobs = find_start_jobs_proc.call(builder_jobs)
      end_jobs = find_end_jobs_proc.call(builder_jobs)

      start_with(*start_jobs) if start_jobs.any?

      end_jobs.each do |end_job|
        end_with end_job, on: :success
      end

      # Define each job
      builder_jobs.each do |job_config|
        job_id = job_config[:id]
        worker_class = job_config[:worker]
        needs_list = job_config[:needs]
        inputs_config = job_config[:inputs]
        condition_proc = job_config[:condition]
        outputs = job_config[:outputs_to_workflow]
        terminates_flag = job_config[:terminates]

        job job_id do
          runs_with worker_class if worker_class

          if needs_list
            needs_array = needs_list.is_a?(Array) ? needs_list : [needs_list]
            needs(*needs_array)
          end

          configure_inputs_proc.call(self, inputs_config)

          if_condition condition_proc if condition_proc

          outputs_to_workflow if outputs || end_jobs.include?(job_id)
          terminates_workflow if terminates_flag || end_jobs.include?(job_id)
        end
      end
    end
  end
end

#build! ⇒ Object

Build and validate in one step



194
195
196
197
 
# File 'lib/fractor/workflow/builder.rb', line 194

def build!
  validate!
  build
end

#clone ⇒ Object

Clone this builder



200
201
202
203
204
205
206
207
 
# File 'lib/fractor/workflow/builder.rb', line 200

def clone
  new_builder = self.class.new(@name)
  new_builder.instance_variable_set(:@input_type_class, @input_type_class)
  new_builder.instance_variable_set(:@output_type_class,
                                    @output_type_class)
  new_builder.instance_variable_set(:@jobs, @jobs.dup)
  new_builder
end

#input_type(klass) ⇒ Object

Set input type for the workflow



28
29
30
31
 
# File 'lib/fractor/workflow/builder.rb', line 28

def input_type(klass)
  @input_type_class = klass
  self
end

#output_type(klass) ⇒ Object

Set output type for the workflow



34
35
36
37
 
# File 'lib/fractor/workflow/builder.rb', line 34

def output_type(klass)
  @output_type_class = klass
  self
end

#remove_job(id) ⇒ Object

Remove a job by id



63
64
65
66
 
# File 'lib/fractor/workflow/builder.rb', line 63

def remove_job(id)
  @jobs.reject! { |j| j[:id] == id }
  self
end

#update_job(id, **options) ⇒ Object

Update a job



69
70
71
72
73
74
75
 
# File 'lib/fractor/workflow/builder.rb', line 69

def update_job(id, **options)
  job = @jobs.find { |j| j[:id] == id }
  return self unless job

  job.merge!(options.compact)
  self
end

#validate! ⇒ Object

Validate the workflow configuration



158
159
160
161
162
163
164
165
166
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
 
# File 'lib/fractor/workflow/builder.rb', line 158

def validate!
  if @name.nil? || @name.empty?
    raise ArgumentError,
          "Workflow must have a name"
  end
  if @jobs.empty?
    raise ArgumentError,
          "Workflow must have at least one job"
  end

  # Check for duplicate job IDs
  job_ids = @jobs.map { |j| j[:id] }
  duplicates = job_ids.select { |id| job_ids.count(id) > 1 }.uniq
  if duplicates.any?
    raise ArgumentError,
          "Duplicate job IDs: #{duplicates.join(', ')}"
  end

  # Check for missing dependencies
  @jobs.each do |job|
    needs = job[:needs]
    next unless needs

    needs_array = needs.is_a?(Array) ? needs : [needs]
    needs_array.each do |dep|
      unless job_ids.include?(dep)
        raise ArgumentError,
              "Job '#{job[:id]}' depends on non-existent job '#{dep}'"
      end
    end
  end

  true
end