Limit Group Migration extractors and loaders to 1 per pipeline

- Restrict defining extractors/loaders to 1 each per pipeline in order
  to reduce `BulkImports::Pipeline::Runner#run` complexity
- If we are going to need to extract/load from/to
  multiple sources/destination we can revisit it and consider
  adding such functionality in the future,
  but there is no need in having it now.

changelogs/unreleased/georgekoltsov-limit-extractors-loaders-to-one.yml

+---
+title: Limit Group Migration extractors and loaders to 1 per pipeline
+merge_request: 50951
+author:
+type: changed

ee/spec/lib/ee/bulk_imports/groups/pipelines/epics_pipeline_spec.rb

@@ -46,13 +46,11 @@ RSpec.describe EE::BulkImports::Groups::Pipelines::EpicsPipeline do
     it { expect(described_class).to include_module(BulkImports::Pipeline::Runner) }
 
     it 'has extractors' do
-      expect(described_class.extractors)
-        .to contain_exactly(
-          {
-            klass: BulkImports::Common::Extractors::GraphqlExtractor,
-            options: {
-              query: EE::BulkImports::Groups::Graphql::GetEpicsQuery
-            }
+      expect(described_class.get_extractor)
+        .to eq(
+          klass: BulkImports::Common::Extractors::GraphqlExtractor,
+          options: {
+            query: EE::BulkImports::Groups::Graphql::GetEpicsQuery
           }
         )
     end
@@ -67,9 +65,7 @@ RSpec.describe EE::BulkImports::Groups::Pipelines::EpicsPipeline do
     end
 
     it 'has loaders' do
-      expect(described_class.loaders).to contain_exactly({
-        klass: EE::BulkImports::Groups::Loaders::EpicsLoader, options: nil
-      })
+      expect(described_class.get_loader).to eq(klass: EE::BulkImports::Groups::Loaders::EpicsLoader, options: nil)
     end
   end
 

lib/bulk_imports/pipeline.rb

@@ -10,16 +10,16 @@ module BulkImports
 
       private
 
-      def extractors
-        @extractors ||= self.class.extractors.map(&method(:instantiate))
+      def extractor
+        @extractor ||= instantiate(self.class.get_extractor)
       end
 
       def transformers
         @transformers ||= self.class.transformers.map(&method(:instantiate))
       end
 
-      def loaders
-        @loaders ||= self.class.loaders.map(&method(:instantiate))
+      def loader
+        @loaders ||= instantiate(self.class.get_loader)
       end
 
       def after_run
@@ -41,7 +41,7 @@ module BulkImports
 
     class_methods do
       def extractor(klass, options = nil)
-        add_attribute(:extractors, klass, options)
+        class_attributes[:extractor] = { klass: klass, options: options }
       end
 
       def transformer(klass, options = nil)
@@ -49,23 +49,23 @@ module BulkImports
       end
 
       def loader(klass, options = nil)
-        add_attribute(:loaders, klass, options)
+        class_attributes[:loader] = { klass: klass, options: options }
       end
 
       def after_run(&block)
         class_attributes[:after_run] = block
       end
 
-      def extractors
-        class_attributes[:extractors]
+      def get_extractor
+        class_attributes[:extractor]
       end
 
       def transformers
         class_attributes[:transformers]
       end
 
-      def loaders
-        class_attributes[:loaders]
+      def get_loader
+        class_attributes[:loader]
       end
 
       def after_run_callback

lib/bulk_imports/pipeline/runner.rb

@@ -12,25 +12,15 @@ module BulkImports
 
         info(context, message: 'Pipeline started', pipeline_class: pipeline)
 
-        extractors.each do |extractor|
-          data = run_pipeline_step(:extractor, extractor.class.name, context) do
-            extractor.extract(context)
+        Array.wrap(extracted_data_from(context)).each do |entry|
+          transformers.each do |transformer|
+            entry = run_pipeline_step(:transformer, transformer.class.name, context) do
+              transformer.transform(context, entry)
+            end
           end
 
-          if data && data.respond_to?(:each)
-            data.each do |entry|
-              transformers.each do |transformer|
-                entry = run_pipeline_step(:transformer, transformer.class.name, context) do
-                  transformer.transform(context, entry)
-                end
-              end
-
-              loaders.each do |loader|
-                run_pipeline_step(:loader, loader.class.name, context) do
-                  loader.load(context, entry)
-                end
-              end
-            end
+          run_pipeline_step(:loader, loader.class.name, context) do
+            loader.load(context, entry)
           end
         end
 
@@ -55,6 +45,12 @@ module BulkImports
         mark_as_failed(context) if abort_on_failure?
       end
 
+      def extracted_data_from(context)
+        run_pipeline_step(:extractor, extractor.class.name, context) do
+          extractor.extract(context)
+        end
+      end
+
       def mark_as_failed(context)
         warn(context, message: 'Pipeline failed', pipeline_class: pipeline)
 

spec/lib/bulk_imports/groups/pipelines/group_pipeline_spec.rb

@@ -75,13 +75,11 @@ RSpec.describe BulkImports::Groups::Pipelines::GroupPipeline do
     it { expect(described_class).to include_module(BulkImports::Pipeline::Runner) }
 
     it 'has extractors' do
-      expect(described_class.extractors)
-        .to contain_exactly(
-          {
-            klass: BulkImports::Common::Extractors::GraphqlExtractor,
-            options: {
-              query: BulkImports::Groups::Graphql::GetGroupQuery
-            }
+      expect(described_class.get_extractor)
+        .to eq(
+          klass: BulkImports::Common::Extractors::GraphqlExtractor,
+          options: {
+            query: BulkImports::Groups::Graphql::GetGroupQuery
           }
         )
     end
@@ -97,9 +95,7 @@ RSpec.describe BulkImports::Groups::Pipelines::GroupPipeline do
     end
 
     it 'has loaders' do
-      expect(described_class.loaders).to contain_exactly({
-        klass: BulkImports::Groups::Loaders::GroupLoader, options: nil
-      })
+      expect(described_class.get_loader).to eq(klass: BulkImports::Groups::Loaders::GroupLoader, options: nil)
     end
   end
 end

spec/lib/bulk_imports/groups/pipelines/subgroup_entities_pipeline_spec.rb

@@ -58,10 +58,7 @@ RSpec.describe BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline do
     it { expect(described_class).to include_module(BulkImports::Pipeline::Runner) }
 
     it 'has extractors' do
-      expect(described_class.extractors).to contain_exactly(
-        klass: BulkImports::Groups::Extractors::SubgroupsExtractor,
-        options: nil
-      )
+      expect(described_class.get_extractor).to eq(klass: BulkImports::Groups::Extractors::SubgroupsExtractor, options: nil)
     end
 
     it 'has transformers' do
@@ -72,10 +69,7 @@ RSpec.describe BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline do
     end
 
     it 'has loaders' do
-      expect(described_class.loaders).to contain_exactly(
-        klass: BulkImports::Common::Loaders::EntityLoader,
-        options: nil
-      )
+      expect(described_class.get_loader).to eq(klass: BulkImports::Common::Loaders::EntityLoader, options: nil)
     end
   end
 end

spec/lib/bulk_imports/pipeline_spec.rb

@@ -24,9 +24,9 @@ RSpec.describe BulkImports::Pipeline do
 
     describe 'getters' do
       it 'retrieves class attributes' do
-        expect(BulkImports::MyPipeline.extractors).to contain_exactly({ klass: BulkImports::Extractor, options: { foo: :bar } })
+        expect(BulkImports::MyPipeline.get_extractor).to eq({ klass: BulkImports::Extractor, options: { foo: :bar } })
         expect(BulkImports::MyPipeline.transformers).to contain_exactly({ klass: BulkImports::Transformer, options: { foo: :bar } })
-        expect(BulkImports::MyPipeline.loaders).to contain_exactly({ klass: BulkImports::Loader, options: { foo: :bar } })
+        expect(BulkImports::MyPipeline.get_loader).to eq({ klass: BulkImports::Loader, options: { foo: :bar } })
         expect(BulkImports::MyPipeline.abort_on_failure?).to eq(true)
       end
     end
@@ -41,20 +41,14 @@ RSpec.describe BulkImports::Pipeline do
         BulkImports::MyPipeline.loader(klass, options)
         BulkImports::MyPipeline.abort_on_failure!
 
-        expect(BulkImports::MyPipeline.extractors)
-          .to contain_exactly(
-            { klass: BulkImports::Extractor, options: { foo: :bar } },
-            { klass: klass, options: options })
+        expect(BulkImports::MyPipeline.get_extractor).to eq({ klass: klass, options: options })
 
         expect(BulkImports::MyPipeline.transformers)
           .to contain_exactly(
             { klass: BulkImports::Transformer, options: { foo: :bar } },
             { klass: klass, options: options })
 
-        expect(BulkImports::MyPipeline.loaders)
-          .to contain_exactly(
-            { klass: BulkImports::Loader, options: { foo: :bar } },
-            { klass: klass, options: options })
+        expect(BulkImports::MyPipeline.get_loader).to eq({ klass: klass, options: options })
 
         expect(BulkImports::MyPipeline.abort_on_failure?).to eq(true)
       end