Class: Kiqchestra::Workflow

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

Overview

The Workflow class provides functionality for orchestrating Sidekiq-based job workflows. It manages task dependencies and tracks their completion status.

Instance Method Summary collapse

Constructor Details

#initialize(workflow_id, metadata) ⇒ Workflow

Initializes the workflow with workflow id and data.

}

Examples:

{

a_job: { deps: [], args: a_job_args },
b_job: { deps: [:a_job], args: b_job_args },
c_job: { deps: [:a_job], args: c_job_args },
d_job: { deps: i[b_job c_job], args: d_job_args }

Parameters:

  • workflow_id (String) —

    Unique ID for the workflow

  • metadata (Hash) —

    Hash defining jobs’ dependencies and arguments



21
22
23
24
25
26
27
28
 
# File 'lib/kiqchestra/workflow.rb', line 21

def initialize(workflow_id, metadata)
  @workflow_id = workflow_id
  @metadata = metadata
  @logger = Logger.new($stdout)

  validate_metadata
  cache_metadata metadata
end

Instance Method Details

#execute ⇒ Object

Starts the workflow execution.



31
32
33
34
35
36
37
38
 
# File 'lib/kiqchestra/workflow.rb', line 31

def execute
  read_progress
  @metadata.each do |job, job_data|
    process_job job, job_data
  end

  conclude_workflow if workflow_complete?
end

#handle_completed_job(job) ⇒ Object

Handles the completion of a job and triggers the next jobs if dependencies are met.

Parameters:

  • job (String) —

    The completed job name (ex. “example_job”)



43
44
45
46
47
48
 
# File 'lib/kiqchestra/workflow.rb', line 43

def handle_completed_job(job)
  update_progress job, "complete"
  log_info "#{job} completed for workflow #{@workflow_id}"

  execute
end