debian-mirror-gitlab/lib/bulk_imports/pipeline.rb

198 lines
4.9 KiB
Ruby
Raw Normal View History

2021-01-29 00:20:46 +05:30
# frozen_string_literal: true
module BulkImports
module Pipeline
extend ActiveSupport::Concern
2021-04-17 20:07:23 +05:30
include Gitlab::Utils::StrongMemoize
2021-02-22 17:27:13 +05:30
include Gitlab::ClassAttributes
2021-03-11 19:13:27 +05:30
include Runner
2021-01-29 00:20:46 +05:30
2021-04-17 20:07:23 +05:30
NotAllowedError = Class.new(StandardError)
2021-09-04 01:27:46 +05:30
ExpiredError = Class.new(StandardError)
FailedError = Class.new(StandardError)
2021-04-17 20:07:23 +05:30
CACHE_KEY_EXPIRATION = 2.hours
2021-09-04 01:27:46 +05:30
NDJSON_EXPORT_TIMEOUT = 30.minutes
2021-04-17 20:07:23 +05:30
2021-03-11 19:13:27 +05:30
def initialize(context)
@context = context
end
2021-02-22 17:27:13 +05:30
2021-04-29 21:17:54 +05:30
def tracker
@tracker ||= context.tracker
end
2021-09-04 01:27:46 +05:30
def portable
@portable ||= context.portable
end
def import_export_config
@import_export_config ||= context.import_export_config
end
def current_user
@current_user ||= context.current_user
end
2021-03-11 19:13:27 +05:30
included do
2021-02-22 17:27:13 +05:30
private
2021-03-11 19:13:27 +05:30
attr_reader :context
2021-04-17 20:07:23 +05:30
# Fetch pipeline extractor.
# An extractor is defined either by instance `#extract(context)` method
# or by using `extractor` DSL.
#
# @example
# class MyPipeline
# extractor MyExtractor, foo: :bar
# end
#
# class MyPipeline
# def extract(context)
# puts 'Fetch some data'
# end
# end
#
# If pipeline implements instance method `extract` - use it
# and ignore class `extractor` method implementation.
2021-03-08 18:12:59 +05:30
def extractor
2021-04-17 20:07:23 +05:30
@extractor ||= self.respond_to?(:extract) ? self : instantiate(self.class.get_extractor)
2021-02-22 17:27:13 +05:30
end
2021-04-17 20:07:23 +05:30
# Fetch pipeline transformers.
#
# A transformer can be defined using:
# - `transformer` class method
# - `transform` instance method
#
# Multiple transformers can be defined within a single
# pipeline and run sequentially for each record in the
# following order:
# - Transformers defined using `transformer` class method
# - Instance method `transform`
#
# Instance method `transform` is always the last to run.
#
# @example
# class MyPipeline
# transformer MyTransformerOne, foo: :bar
# transformer MyTransformerTwo, foo: :bar
#
# def transform(context, data)
# # perform transformation here
# end
# end
#
# In the example above `#transform` is the first to run and
# `MyTransformerTwo` method is the last.
2021-02-22 17:27:13 +05:30
def transformers
2021-04-17 20:07:23 +05:30
strong_memoize(:transformers) do
defined_transformers = self.class.transformers.map(&method(:instantiate))
transformers = []
transformers << self if respond_to?(:transform)
transformers.concat(defined_transformers)
transformers
end
2021-02-22 17:27:13 +05:30
end
2021-04-17 20:07:23 +05:30
# Fetch pipeline loader.
# A loader is defined either by instance method `#load(context, data)`
# or by using `loader` DSL.
#
# @example
# class MyPipeline
# loader MyLoader, foo: :bar
# end
#
# class MyPipeline
# def load(context, data)
# puts 'Load some data'
# end
# end
#
# If pipeline implements instance method `load` - use it
# and ignore class `loader` method implementation.
2021-03-08 18:12:59 +05:30
def loader
2021-04-17 20:07:23 +05:30
@loader ||= self.respond_to?(:load) ? self : instantiate(self.class.get_loader)
2021-02-22 17:27:13 +05:30
end
def pipeline
@pipeline ||= self.class.name
end
def instantiate(class_config)
2021-04-17 20:07:23 +05:30
options = class_config[:options]
if options
2021-09-04 01:27:46 +05:30
class_config[:klass].new(**class_config[:options])
2021-04-17 20:07:23 +05:30
else
class_config[:klass].new
end
2021-02-22 17:27:13 +05:30
end
def abort_on_failure?
self.class.abort_on_failure?
end
end
class_methods do
def extractor(klass, options = nil)
2021-03-08 18:12:59 +05:30
class_attributes[:extractor] = { klass: klass, options: options }
2021-02-22 17:27:13 +05:30
end
def transformer(klass, options = nil)
add_attribute(:transformers, klass, options)
end
def loader(klass, options = nil)
2021-03-08 18:12:59 +05:30
class_attributes[:loader] = { klass: klass, options: options }
2021-02-22 17:27:13 +05:30
end
2021-03-08 18:12:59 +05:30
def get_extractor
class_attributes[:extractor]
2021-02-22 17:27:13 +05:30
end
def transformers
2021-04-17 20:07:23 +05:30
class_attributes[:transformers] || []
2021-02-22 17:27:13 +05:30
end
2021-03-08 18:12:59 +05:30
def get_loader
class_attributes[:loader]
2021-02-22 17:27:13 +05:30
end
def abort_on_failure!
class_attributes[:abort_on_failure] = true
end
def abort_on_failure?
class_attributes[:abort_on_failure]
end
2021-09-04 01:27:46 +05:30
def ndjson_pipeline!
class_attributes[:ndjson_pipeline] = true
end
def ndjson_pipeline?
class_attributes[:ndjson_pipeline]
end
def relation_name(name)
class_attributes[:relation_name] = name
end
def relation
class_attributes[:relation_name]
end
2021-02-22 17:27:13 +05:30
private
def add_attribute(sym, klass, options)
class_attributes[sym] ||= []
class_attributes[sym] << { klass: klass, options: options }
end
2021-01-29 00:20:46 +05:30
end
end
end