Module: Webhookdb::Postgres::ModelUtilities::DatasetMethods

Defined in:
lib/webhookdb/postgres/model_utilities.rb

Instance Method Summary collapse

Instance Method Details

#each_cursor_page(page_size: 500, order: :id, &block) ⇒ Object

Call a block for each row in a dataset. This is the same as paged_each or use_cursor.each, except that for each page, rows are re-fetched using self.where(primary_key => [pks]).all to enable eager loading.

(Note that paged_each does not do eager loading, which makes enumerating model associations very slow)

Raises:

  • (LocalJumpError) 


362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
 
# File 'lib/webhookdb/postgres/model_utilities.rb', line 362

def each_cursor_page(page_size: 500, order: :id, &block)
  raise LocalJumpError unless block
  raise "dataset requires a use_cursor method, class may need `extension(:pagination)`" unless
    self.respond_to?(:use_cursor)
  model = self.model
  pk = model.primary_key
  current_chunk_pks = []
  order = [order] unless order.respond_to?(:to_ary)
  self.naked.select(pk).order(*order).use_cursor(rows_per_fetch: page_size, hold: true).each do |row|
    current_chunk_pks << row[pk]
    next if current_chunk_pks.length < page_size
    page = model.where(pk => current_chunk_pks).order(*order).all
    current_chunk_pks.clear
    page.each(&block)
  end
  model.where(pk => current_chunk_pks).order(*order).all.each(&block)
end

#each_cursor_page_action(action:, page_size: 500, order: :id) ⇒ Object

See each_cursor_page, but takes an additional action on each chunk of returned rows. The action is called with pages of return values from the block when a page is is reached. Each call to action should return nil, a result, or an array of results (nil results are ignored).

The most common case is for ETL: process one dataset, map it in a block to return new row values, and multi_insert it into a different table.

Raises:

  • (LocalJumpError) 


386
387
388
389
390
391
392
393
394
395
396
397
398
399
 
# File 'lib/webhookdb/postgres/model_utilities.rb', line 386

def each_cursor_page_action(action:, page_size: 500, order: :id)
  raise LocalJumpError unless block_given?
  returned_rows_chunk = []
  self.each_cursor_page(page_size:, order:) do |instance|
    new_row = yield(instance)
    next if action.nil? || new_row.nil?
    new_row.respond_to?(:to_ary) ? returned_rows_chunk.concat(new_row) : returned_rows_chunk.push(new_row)
    if returned_rows_chunk.length >= page_size
      action.call(returned_rows_chunk)
      returned_rows_chunk.clear
    end
  end
  action&.call(returned_rows_chunk)
end

#reduce_expr(op_symbol, operands, method: :where) ⇒ Object

Helper for applying multiple conditions for Sequel, where some can be nil.



353
354
355
 
# File 'lib/webhookdb/postgres/model_utilities.rb', line 353

def reduce_expr(op_symbol, operands, method: :where)
  return Webhookdb::Dbutil.reduce_expr(self, op_symbol, operands, method:)
end