Module: JobWorkflow::DSL::ClassMethods

Included in:
JobWorkflow::DSL
Defined in:
lib/job_workflow/dsl.rb,
sig/generated/job_workflow/dsl.rbs

Instance Method Summary collapse

Instance Method Details

#_workflow ⇒ Workflow

Returns:



49
 
# File 'sig/generated/job_workflow/dsl.rbs', line 49

def _workflow: () -> Workflow

#after(*task_names) {|arg0| ... } ⇒ void

This method returns an undefined value.

: (*Symbol) { (Context) -> void } -> void

Parameters:

  • (Symbol)

Yields:

Yield Parameters:

Yield Returns:

  • (void) 


172
173
174
 
# File 'lib/job_workflow/dsl.rb', line 172

def after(*task_names, &block)
  _workflow.add_hook(:after, task_names:, block:)
end

#argument(argument_name, type, default: nil) ⇒ void

This method returns an undefined value.

: (Symbol argument_name, String type, ?default: untyped) -> void

Parameters:

  • argument_name (Symbol)
  • type (String)
  • default: (Object) (defaults to: nil) 


114
115
116
 
# File 'lib/job_workflow/dsl.rb', line 114

def argument(argument_name, type, default: nil)
  _workflow.add_argument(ArgumentDef.new(name: argument_name, type:, default:))
end

#around(*task_names) {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

: (*Symbol) { (Context, TaskCallable) -> void } -> void

Parameters:

  • (Symbol)

Yields:

Yield Parameters:

Yield Returns:

  • (void) 


177
178
179
 
# File 'lib/job_workflow/dsl.rb', line 177

def around(*task_names, &block)
  _workflow.add_hook(:around, task_names:, block:)
end

#before(*task_names) {|arg0| ... } ⇒ void

This method returns an undefined value.

: (*Symbol) { (Context) -> void } -> void

Parameters:

  • (Symbol)

Yields:

Yield Parameters:

Yield Returns:

  • (void) 


167
168
169
 
# File 'lib/job_workflow/dsl.rb', line 167

def before(*task_names, &block)
  _workflow.add_hook(:before, task_names:, block:)
end

#class_attribute ⇒ void

This method returns an undefined value.

Parameters:

  • (Symbol)
  • default: (Object) 


47
 
# File 'sig/generated/job_workflow/dsl.rbs', line 47

def class_attribute: (Symbol, default: untyped) -> void

#dry_run(value = nil, &block) ⇒ void

This method returns an undefined value.

: (?bool) ?{ (Context) -> bool } -> void

Parameters:

  • (Boolean) 


233
234
235
 
# File 'lib/job_workflow/dsl.rb', line 233

def dry_run(value = nil, &block)
  _workflow.dry_run_config = block || value
end

#enqueue ⇒ void

This method returns an undefined value.

Parameters:

  • (Hash[untyped, untyped]) 


55
 
# File 'sig/generated/job_workflow/dsl.rbs', line 55

def enqueue: (Hash[untyped, untyped]) -> void

#from_context(context) ⇒ DSL

: (Context) -> DSL

Parameters:

Returns:



103
104
105
106
107
108
109
110
111
 
# File 'lib/job_workflow/dsl.rb', line 103

def from_context(context) # rubocop:disable Metrics/AbcSize
  new_context = context.dup
  task = new_context._task_context.task
  job = new(new_context.arguments.to_h)
  new_context._job = job
  job._context = new_context
  job.set(queue: task.enqueue.queue) if !task.nil? && !task.enqueue.queue.nil?
  job
end

#limits_concurrency ⇒ void

This method returns an undefined value.

Parameters:

  • to: (Integer)
  • key: (^(untyped) -> untyped)
  • duration: (ActiveSupport::Duration, nil)
  • group: (String, nil)
  • on_conflict: (Symbol, nil) 


61
 
# File 'sig/generated/job_workflow/dsl.rbs', line 61

def limits_concurrency: (to: Integer, key: ^(untyped) -> untyped, ?duration: ActiveSupport::Duration?, ?group: String?, ?on_conflict: Symbol?) -> void

#name ⇒ String

Returns:

  • (String) 


53
 
# File 'sig/generated/job_workflow/dsl.rbs', line 53

def name: () -> String

#new ⇒ DSL

Parameters:

  • (Hash[untyped, untyped])

Returns:



51
 
# File 'sig/generated/job_workflow/dsl.rbs', line 51

def new: (Hash[untyped, untyped]) -> DSL

#on_error(*task_names) {|arg0, arg1, arg2| ... } ⇒ void

This method returns an undefined value.

: (*Symbol) { (Context, StandardError, Task) -> void } -> void

Parameters:

  • (Symbol)

Yields:

Yield Parameters:

Yield Returns:

  • (void) 


182
183
184
 
# File 'lib/job_workflow/dsl.rb', line 182

def on_error(*task_names, &block)
  _workflow.add_hook(:error, task_names:, block:)
end

#queue_as ⇒ String

Returns:

  • (String) 


59
 
# File 'sig/generated/job_workflow/dsl.rbs', line 59

def queue_as: () -> String

#queue_name ⇒ String

Returns:

  • (String) 


57
 
# File 'sig/generated/job_workflow/dsl.rbs', line 57

def queue_name: () -> String

#schedule(expression, key: nil, queue: nil, priority: nil, args: {}, description: nil) ⇒ void

This method returns an undefined value.

rubocop:disable Metrics/ParameterLists : ( String expression, ?key: (String | Symbol)?, ?queue: String?, ?priority: Integer?, ?args: Hash[Symbol, untyped], ?description: String? ) -> void

Parameters:

  • expression (String)
  • key: (String, Symbol, nil) (defaults to: nil)
  • queue: (String, nil) (defaults to: nil)
  • priority: (Integer, nil) (defaults to: nil)
  • args: (Hash[Symbol, untyped]) (defaults to: {})
  • description: (String, nil) (defaults to: nil) 


246
247
248
249
250
251
252
253
254
255
256
257
258
 
# File 'lib/job_workflow/dsl.rb', line 246

def schedule(expression, key: nil, queue: nil, priority: nil, args: {}, description: nil)
  _workflow.add_schedule(
    Schedule.new(
      expression:,
      class_name: name,
      key:,
      queue:,
      priority:,
      args:,
      description:
    )
  )
end

#task(task_name, each: Task::DEFAULT_EACH, enqueue: nil, retry: 0, output: {}, depends_on: [], condition: ->(_ctx) { true }, throttle: {}, timeout: nil, dependency_wait: {}, dry_run: false) {|arg0| ... } ⇒ void

This method returns an undefined value.

rubocop:disable Metrics/ParameterLists

: ( Symbol task_name, ?each: ^(Context) -> untyped, ?enqueue: true | false | ^(Context) -> bool | Hash[Symbol, untyped], ?retry: Integer | Hash[Symbol, untyped], ?output: Hash[Symbol, String], ?depends_on: Array, ?condition: ^(Context) -> bool, ?throttle: Integer | Hash[Symbol, untyped], ?timeout: Numeric?, ?dependency_wait: Hash[Symbol, untyped], ?dry_run: bool | ^(Context) -> bool ) { (untyped) -> void } -> void

Parameters:

  • task_name (Symbol)
  • each: (^(Context) -> untyped) (defaults to: Task::DEFAULT_EACH)
  • enqueue: (true, false, ^(Context) -> bool, Hash[Symbol, untyped]) (defaults to: nil)
  • retry: (Integer, Hash[Symbol, untyped]) (defaults to: 0)
  • output: (Hash[Symbol, String]) (defaults to: {})
  • depends_on: (Array[Symbol]) (defaults to: [])
  • condition: (^(Context) -> bool) (defaults to: ->(_ctx) { true })
  • throttle: (Integer, Hash[Symbol, untyped]) (defaults to: {})
  • timeout: (Numeric, nil) (defaults to: nil)
  • dependency_wait: (Hash[Symbol, untyped]) (defaults to: {})
  • dry_run: (bool, ^(Context) -> bool) (defaults to: false)

Yields:

Yield Parameters:

  • arg0 (Object)

Yield Returns:

  • (void) 


133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/job_workflow/dsl.rb', line 133

def task(
  task_name,
  each: Task::DEFAULT_EACH,
  enqueue: nil,
  retry: 0,
  output: {},
  depends_on: [],
  condition: ->(_ctx) { true },
  throttle: {},
  timeout: nil,
  dependency_wait: {},
  dry_run: false,
  &block
)
  new_task = Task.new(
    job_name: name,
    name: task_name,
    block: block,
    enqueue:,
    each:,
    task_retry: binding.local_variable_get(:retry),
    output:,
    depends_on:,
    condition:,
    throttle:,
    timeout:,
    dependency_wait:,
    dry_run:
  )
  _workflow.add_task(new_task)
end

#workflow_concurrency(to:, key:, **opts) ⇒ void

This method returns an undefined value.

Configures concurrency limits for this workflow job.

Unlike limits_concurrency (SolidQueue's raw API), this method passes a Context as the first argument to the key Proc, giving access to workflow-aware information such as arguments, sub_job?, and concurrency_key.

When _context is not yet initialized (e.g. during enqueue before perform), a temporary Context is built from the job's arguments so the key Proc can always rely on ctx.arguments.

: ( to: Integer, key: ^(Context) -> String?, ?duration: ActiveSupport::Duration?, ?group: String?, ?on_conflict: Symbol? ) -> void

Examples:

Limit duplicate workflow runs by argument

workflow_concurrency to: 1,
  key: ->(ctx) { "my_job:#{ctx.arguments.tenant_id}" },
  on_conflict: :discard

Separate parent and sub-job concurrency keys

workflow_concurrency to: 1,
  key: ->(ctx) {
    ctx.sub_job? ? ctx.concurrency_key : "my_job:#{ctx.arguments.name}"
  },
  on_conflict: :discard

Parameters:

  • to: (Integer)
  • key: (^(Context) -> String, nil)
  • duration: (ActiveSupport::Duration, nil)
  • group: (String, nil)
  • on_conflict: (Symbol, nil) 


216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
 
# File 'lib/job_workflow/dsl.rb', line 216

def workflow_concurrency(to:, key:, **opts)
  concurrency_key_proc = key
  wrapped_key = proc {
    # @type self: DSL
    ctx = _context || Context.from_hash(
      job: self, workflow: self.class._workflow
    )._update_arguments((arguments.first || {}).symbolize_keys)
    concurrency_key_proc.call(ctx)
  } #: ^(untyped) -> untyped
  limits_concurrency(
    to:,
    key: wrapped_key,
    **opts
  )
end