Class: ActiveJob::JobsRelation

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/active_job/jobs_relation.rb

Overview

A relation of jobs that can be filtered and acted on.

Relations of jobs are normally fetched via ActiveJob.jobs or through a given queue (+ActiveJob::Queue#jobs+).

This class offers a fluid interface to query a subset of jobs. For example:

queue = ActiveJob.queues[:default]
queue.jobs.limit(10).where(job_class_name: "DummyJob").last

Relations are enumerable, so you can use Enumerable methods on them. Notice however that using these methods will imply loading all the relation in memory, which could introduce performance concerns.

Internally, ActiveJob will always use paginated queries to the underlying queue adapter. The page size can be controlled via config.active_job.default_page_size (1000 by default).

There are additional performance concerns depending on the configured adapter. Please check ActiveJob::Relation#where, ActiveJob::Relation#count.

Constant Summary collapse

STATUSES = 
%i[pending failed in_progress blocked scheduled finished]
FILTERS = 
%i[queue_name job_class_name]
PROPERTIES = 
%i[queue_name status offset_value limit_value job_class_name worker_id recurring_task_id finished_at]
ALL_JOBS_LIMIT =

When no limit value it defaults to "all jobs"

100_000_000

Instance Method Summary collapse

Constructor Details

#initialize(queue_adapter: ActiveJob::Base.queue_adapter, default_page_size: ActiveJob::Base.default_page_size) ⇒ JobsRelation

Returns a new instance of JobsRelation.



36
37
38
39
40
41
 
# File 'lib/active_job/jobs_relation.rb', line 36

def initialize(queue_adapter: ActiveJob::Base.queue_adapter, default_page_size: ActiveJob::Base.default_page_size)
  @queue_adapter = queue_adapter
  @default_page_size = default_page_size

  set_defaults
end

Instance Method Details

#count ⇒ Object Also known as: length, size

Returns the number of jobs in the relation.

When filtering jobs, if the adapter doesn't support the filter(s) directly, this will load all the jobs in memory to filter them.



98
99
100
101
102
103
104
 
# File 'lib/active_job/jobs_relation.rb', line 98

def count
  if loaded? || filtering_needed?
    to_a.length
  else
    query_count
  end
end

#discard_all ⇒ Object

Discard all the jobs in the relation.



147
148
149
150
 
# File 'lib/active_job/jobs_relation.rb', line 147

def discard_all
  queue_adapter.discard_all_jobs(self)
  nil
end

#discard_job(job) ⇒ Object

Discard the provided job.



153
154
155
 
# File 'lib/active_job/jobs_relation.rb', line 153

def discard_job(job)
  queue_adapter.discard_job(job, self)
end

#dispatch_job(job) ⇒ Object

Dispatch the provided job.

This operation is only valid for blocked or scheduled jobs. It will raise an error ActiveJob::Errors::InvalidOperation otherwise.

Raises:



161
162
163
164
165
 
# File 'lib/active_job/jobs_relation.rb', line 161

def dispatch_job(job)
  raise ActiveJob::Errors::InvalidOperation, "This operation can only be performed on blocked or scheduled jobs, but this job is #{job.status}" unless job.blocked? || job.scheduled?

  queue_adapter.dispatch_job(job, self)
end

#each(&block) ⇒ Object 



123
124
125
 
# File 'lib/active_job/jobs_relation.rb', line 123

def each(&block)
  loaded_jobs&.each(&block) || load_jobs(&block)
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


109
110
111
 
# File 'lib/active_job/jobs_relation.rb', line 109

def empty?
  count == 0
end

#filtering_needed? ⇒ Boolean

Returns:

  • (Boolean) 


215
216
217
 
# File 'lib/active_job/jobs_relation.rb', line 215

def filtering_needed?
  filters.any?
end

#find_by_id(job_id) ⇒ Object

Find a job by id.

Returns nil when not found.



170
171
172
 
# File 'lib/active_job/jobs_relation.rb', line 170

def find_by_id(job_id)
  queue_adapter.find_job(job_id, self)
end

#find_by_id!(job_id) ⇒ Object

Find a job by id.

Raises ActiveJob::Errors::JobNotFoundError when not found.



177
178
179
 
# File 'lib/active_job/jobs_relation.rb', line 177

def find_by_id!(job_id)
  queue_adapter.find_job(job_id, self) or raise ActiveJob::Errors::JobNotFoundError.new(job_id, self)
end

#in_batches(of: default_page_size, order: :asc, &block) ⇒ Object 



194
195
196
197
198
199
200
201
202
203
204
205
 
# File 'lib/active_job/jobs_relation.rb', line 194

def in_batches(of: default_page_size, order: :asc, &block)
  validate_looping_in_batches_is_possible

  case order
  when :asc
    in_ascending_batches(of: of, &block)
  when :desc
    in_descending_batches(of: of, &block)
  else
    raise "Unsupported order: #{order}. Valid values: :asc, :desc."
  end
end

#job_class_names(from_first: 500) ⇒ Object

Returns an array of jobs class names in the first from_first jobs.



182
183
184
 
# File 'lib/active_job/jobs_relation.rb', line 182

def job_class_names(from_first: 500)
  first(from_first).collect(&:job_class_name).uniq
end

#limit(limit) ⇒ Object

Sets the max number of jobs to fetch in the query.



90
91
92
 
# File 'lib/active_job/jobs_relation.rb', line 90

def limit(limit)
  clone_with limit_value: limit
end

#limit_value_provided? ⇒ Boolean

Returns:

  • (Boolean) 


211
212
213
 
# File 'lib/active_job/jobs_relation.rb', line 211

def limit_value_provided?
  limit_value.present? && limit_value != ActiveJob::JobsRelation::ALL_JOBS_LIMIT
end

#offset(offset) ⇒ Object

Sets an offset for the jobs-fetching query. The first position is 0.



85
86
87
 
# File 'lib/active_job/jobs_relation.rb', line 85

def offset(offset)
  clone_with offset_value: offset
end

#paginated? ⇒ Boolean

Returns:

  • (Boolean) 


207
208
209
 
# File 'lib/active_job/jobs_relation.rb', line 207

def paginated?
  offset_value > 0 || limit_value_provided?
end

#reload ⇒ Object 



186
187
188
189
190
191
192
 
# File 'lib/active_job/jobs_relation.rb', line 186

def reload
  @count = nil
  @loaded_jobs = nil
  @filters = nil

  self
end

#retry_all ⇒ Object

Retry all the jobs in the queue.

This operation is only valid for sets of failed jobs. It will raise an error ActiveJob::Errors::InvalidOperation otherwise.



131
132
133
134
135
 
# File 'lib/active_job/jobs_relation.rb', line 131

def retry_all
  ensure_failed_status
  queue_adapter.retry_all_jobs(self)
  nil
end

#retry_job(job) ⇒ Object

Retry the provided job.

This operation is only valid for sets of failed jobs. It will raise an error ActiveJob::Errors::InvalidOperation otherwise.



141
142
143
144
 
# File 'lib/active_job/jobs_relation.rb', line 141

def retry_job(job)
  ensure_failed_status
  queue_adapter.retry_job(job, self)
end

#to_s ⇒ Object Also known as: inspect 



113
114
115
116
117
118
119
 
# File 'lib/active_job/jobs_relation.rb', line 113

def to_s
  properties_with_values = PROPERTIES.collect do |name|
    value = public_send(name)
    "#{name}: #{value}" unless value.nil?
  end.compact.join(", ")
  "<Jobs with [#{properties_with_values}]> (loaded: #{loaded?})"
end

#where(job_class_name: nil, queue_name: nil, worker_id: nil, recurring_task_id: nil, finished_at: nil) ⇒ Object

Returns a ActiveJob::JobsRelation with the configured filtering options.

Options

  • :job_class_name - To only include the jobs of a given class. Depending on the configured queue adapter, this will perform the filtering in memory, which could introduce performance concerns for large sets of jobs.
  • :queue_name - To only include the jobs in the provided queue.
  • :worker_id - To only include the jobs processed by the provided worker.
  • :recurring_task_id - To only include the jobs corresponding to runs of a recurring task.
  • :finished_at - (Range) To only include the jobs finished between the provided range


55
56
57
58
59
60
61
62
63
64
 
# File 'lib/active_job/jobs_relation.rb', line 55

def where(job_class_name: nil, queue_name: nil, worker_id: nil, recurring_task_id: nil, finished_at: nil)
  # Remove nil arguments to avoid overriding parameters when concatenating +where+ clauses
  arguments = {job_class_name: job_class_name,
               queue_name: queue_name&.to_s,
               worker_id: worker_id,
               recurring_task_id: recurring_task_id,
               finished_at: finished_at}.compact

  clone_with(**arguments)
end

#with_status(status) ⇒ Object 



66
67
68
69
70
71
72
 
# File 'lib/active_job/jobs_relation.rb', line 66

def with_status(status)
  if status.to_sym.in? STATUSES
    clone_with status: status.to_sym
  else
    self
  end
end