Class: Fractor::Workflow::WorkflowContext

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

Overview

Manages data flow and state during workflow execution. Stores workflow inputs, job outputs, and provides resolution of data dependencies.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(workflow_input, correlation_id: nil, logger: nil) ⇒ WorkflowContext

Returns a new instance of WorkflowContext.



10
11
12
13
14
15
 
# File 'lib/fractor/workflow/workflow_context.rb', line 10

def initialize(workflow_input, correlation_id: nil, logger: nil)
  @workflow_input = workflow_input
  @job_outputs = {}
  @correlation_id = correlation_id || generate_correlation_id
  @logger = logger || WorkflowLogger.new(correlation_id: @correlation_id)
end

Instance Attribute Details

#correlation_id ⇒ Object (readonly)

Returns the value of attribute correlation_id.



8
9
10
 
# File 'lib/fractor/workflow/workflow_context.rb', line 8

def correlation_id
  @correlation_id
end

#job_outputs ⇒ Object (readonly)

Returns the value of attribute job_outputs.



8
9
10
 
# File 'lib/fractor/workflow/workflow_context.rb', line 8

def job_outputs
  @job_outputs
end

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



8
9
10
 
# File 'lib/fractor/workflow/workflow_context.rb', line 8

def logger
  @logger
end

#workflow_input ⇒ Object (readonly)

Returns the value of attribute workflow_input.



8
9
10
 
# File 'lib/fractor/workflow/workflow_context.rb', line 8

def workflow_input
  @workflow_input
end

Instance Method Details

#build_job_input(job) ⇒ Lutaml::Model::Serializable

Build input for a job based on its input mappings.

Parameters:

  • job (Job) —

    The job to build input for

Returns:

  • (Lutaml::Model::Serializable) —

    The constructed input



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/fractor/workflow/workflow_context.rb', line 37

def build_job_input(job)
  return @workflow_input if job.input_mappings[:workflow]

  input_type = job.input_type
  unless input_type
    raise "Job '#{job.name}' has no input_type defined in its worker"
  end

  # Collect attributes from all mapped sources
  input_attrs = {}

  job.input_mappings.each do |source_job_name, attr_mappings|
    source_output = job_output(source_job_name)
    unless source_output
      raise "Job '#{job.name}' depends on '#{source_job_name}' but its output is not available"
    end

    if attr_mappings == :all
      # Map all attributes from source to input
      copy_all_attributes(source_output, input_attrs, input_type)
    else
      # Map specific attributes
      attr_mappings.each do |target_attr, source_attr|
        target_attr = target_attr.to_sym
        source_attr = source_attr.to_sym

        # Get value from source output
        value = if source_output.respond_to?(source_attr)
                  source_output.send(source_attr)
                elsif source_output.respond_to?(:[])
                  source_output[source_attr]
                else
                  raise "Source output from '#{source_job_name}' does not have attribute '#{source_attr}'"
                end

        input_attrs[target_attr] = value
      end
    end
  end

  # Create input instance
  input_type.new(**input_attrs)
end

#job_completed?(job_name) ⇒ Boolean

Check if a job's output is available.

Parameters:

  • job_name (String) —

    The job name

Returns:

  • (Boolean) —

    Whether the output is available



85
86
87
 
# File 'lib/fractor/workflow/workflow_context.rb', line 85

def job_completed?(job_name)
  @job_outputs.key?(job_name)
end

#job_output(job_name) ⇒ Lutaml::Model::Serializable?

Get the output of a completed job.

Parameters:

  • job_name (String) —

    The job name

Returns:

  • (Lutaml::Model::Serializable, nil) —

    The job's output



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

def job_output(job_name)
  @job_outputs[job_name]
end

#store_job_output(job_name, output) ⇒ Object

Store the output of a completed job.

Parameters:

  • job_name (String) —

    The job name

  • output (Lutaml::Model::Serializable) —

    The job's output



21
22
23
 
# File 'lib/fractor/workflow/workflow_context.rb', line 21

def store_job_output(job_name, output)
  @job_outputs[job_name] = output
end

#to_h ⇒ Object

Convert context to hash for debugging/logging



90
91
92
93
94
95
96
 
# File 'lib/fractor/workflow/workflow_context.rb', line 90

def to_h
  {
    correlation_id: @correlation_id,
    workflow_input: @workflow_input.class.name,
    completed_jobs: @job_outputs.keys,
  }
end