refactor: EtlPipeline entry via Create() instance seed (Option 2)

CHANGELOG.md

@@ -28,8 +28,10 @@ Minor release: adds a generic, format-agnostic ETL pipeline. No breaking change.
 - `EtlPipeline` — a generic, format-agnostic pipeline that composes any source, transformer
   stages, and a loader into a single runnable flow, complementing the existing fluent
   `Pipeline` (Extract/Transform/Load) builder:
-  - `EtlPipeline.From(IAsyncEnumerable<T>)` and `From(ExtractorBase<T, TProgress>)` factories,
-    plus the `EtlPipeline.Source` sentinel that format packages extend with source factories.
+  - `EtlPipeline.Create()` returns a fresh builder seed; `From(IAsyncEnumerable<T>)` and
+    `From(ExtractorBase<T, TProgress>)` factories start the chain from any source. Format
+    packages extend the `EtlPipeline` instance with class-named source factories, e.g.
+    `EtlPipeline.Create().CsvExtractor<Order>("orders.csv")`.
   - `IEtlPipeline<T>` with `Through` (four overloads — an `ITransformAsync<T, TOut>` or
     `ITransformWithCancellationAsync<T, TOut>` transformer, or a stream-to-stream delegate,
     with or without a `CancellationToken`), `To<TProgress>(LoaderBase<T, TProgress>)`, and

examples/Net4.8/Example8-EtlPipeline/Program.cs

@@ -27,7 +27,9 @@ private static async Task Main()
             // Each Through returns IEtlPipeline<TOut>, so the element type flows
             // string -> int -> double -> string across the chain and the compiler enforces
             // that each stage's output matches the next stage's input.
-            await EtlPipeline.From(RawNumbers())
+            await EtlPipeline
+                .Create()
+                .From(RawNumbers())
                 .Through(new ParseIntTransformer())   // string -> int
                 .Through(new DoubleTransformer())     // int    -> double
                 .Through(new FormatTransformer())     // double -> string
@@ -38,8 +40,8 @@ await EtlPipeline.From(RawNumbers())
         }
 
 
-        // Any IAsyncEnumerable<T> is a valid source via EtlPipeline.From(...).
-        // An ExtractorBase<T, TProgress> works too: EtlPipeline.From(myExtractor).
+        // Any IAsyncEnumerable<T> is a valid source via EtlPipeline.Create().From(...).
+        // An ExtractorBase<T, TProgress> works too: EtlPipeline.Create().From(myExtractor).
         private static async IAsyncEnumerable<string> RawNumbers()
         {
             for (var i = 1; i <= 8; i++)

examples/Net8.0/Example8-EtlPipeline/Program.cs

@@ -24,7 +24,9 @@ private static async Task Main()
         // Each Through returns IEtlPipeline<TOut>, so the element type flows
         // string -> int -> double -> string across the chain and the compiler enforces
         // that each stage's output matches the next stage's input.
-        await EtlPipeline.From(RawNumbers())
+        await EtlPipeline
+            .Create()
+            .From(RawNumbers())
             .Through(new ParseIntTransformer())   // string -> int
             .Through(new DoubleTransformer())     // int    -> double
             .Through(new FormatTransformer())     // double -> string
@@ -35,8 +37,8 @@ await EtlPipeline.From(RawNumbers())
     }
 
 
-    // Any IAsyncEnumerable<T> is a valid source via EtlPipeline.From(...).
-    // An ExtractorBase<T, TProgress> works too: EtlPipeline.From(myExtractor).
+    // Any IAsyncEnumerable<T> is a valid source via EtlPipeline.Create().From(...).
+    // An ExtractorBase<T, TProgress> works too: EtlPipeline.Create().From(myExtractor).
     private static async IAsyncEnumerable<string> RawNumbers()
     {
         for (var i = 1; i <= 8; i++)

src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipeline.cs

@@ -6,10 +6,11 @@
 namespace Wolfgang.Etl.Abstractions;
 
 /// <summary>
-/// Entry point for building a generic, format-agnostic ETL pipeline. Start from a source — either a
-/// built-in <see cref="From{T}(IAsyncEnumerable{T})"/> / <see cref="From{T, TProgress}(ExtractorBase{T, TProgress})"/>
-/// factory, or a format-specific factory hung off <see cref="Source"/> by a format package — chain
-/// append transformer stages on <see cref="IEtlPipeline{T}"/>, terminate with a sink, then call
+/// Entry point for building a generic, format-agnostic ETL pipeline. Obtain a fresh builder with
+/// <see cref="Create"/>, start from a source — either a built-in <see cref="From{T}(IAsyncEnumerable{T})"/> /
+/// <see cref="From{T, TProgress}(ExtractorBase{T, TProgress})"/> factory, or a format-specific factory
+/// hung off the <see cref="EtlPipeline"/> instance by a format package — chain append transformer stages
+/// on <see cref="IEtlPipeline{T}"/>, terminate with a sink, then call
 /// <see cref="IEtlPipelineSink.RunAsync(IProgress{EtlPipelineProgress}, CancellationToken)"/>.
 /// </summary>
 /// <remarks>
@@ -19,6 +20,13 @@ namespace Wolfgang.Etl.Abstractions;
 /// <c>System.IO.Pipelines</c>.
 /// </para>
 /// <para>
+/// <see cref="Create"/> returns a fresh <see cref="EtlPipeline"/> seed: it carries no data itself and
+/// exists to give source factories a strongly-typed receiver. Format packages extend the instance with
+/// class-named factories — for example a CSV package adds
+/// <c>public static ICsvExtractorBuilder&lt;T&gt; CsvExtractor&lt;T&gt;(this EtlPipeline pipeline, string path)</c>,
+/// enabling <c>EtlPipeline.Create().CsvExtractor&lt;Order&gt;("orders.csv")</c>.
+/// </para>
+/// <para>
 /// The core exposes only the plumbing — a source, <see cref="IEtlPipeline{T}.Through{TOut}(ITransformAsync{T, TOut})"/>
 /// for appending transformer stages, and a sink. The LINQ-flavored operators (<c>Where</c>,
 /// <c>Select</c>, <c>Distinct</c>, …) are extension methods shipped by <c>Wolfgang.Etl.Transformers</c>,
@@ -27,19 +35,25 @@ namespace Wolfgang.Etl.Abstractions;
 /// </remarks>
 /// <example>
 /// <code>
-/// await EtlPipeline.From(records)
+/// await EtlPipeline.Create()
+///     .From(records)
 ///     .Through(new WhereTransformer&lt;Order&gt;(r =&gt; r.Amount &gt; 0))
 ///     .To(sqlLoader)
 ///     .RunAsync(progress, token);
 /// </code>
 /// </example>
-public static class EtlPipeline
+public sealed class EtlPipeline
 {
+    private EtlPipeline()
+    {
+    }
+
+
     /// <summary>
-    /// The sentinel that format packages extend with source factories, enabling the
-    /// <c>EtlPipeline.Source.CsvExtractor&lt;T&gt;(...)</c> shape. See <see cref="EtlPipelineSource"/>.
+    /// Creates a new, empty <see cref="EtlPipeline"/> builder to start a fluent pipeline chain.
     /// </summary>
-    public static EtlPipelineSource Source { get; } = new();
+    /// <returns>A fresh builder instance.</returns>
+    public static EtlPipeline Create() => new();
 
 
     /// <summary>
@@ -50,7 +64,7 @@ public static class EtlPipeline
     /// <param name="source">The stream that seeds the pipeline.</param>
     /// <returns>An <see cref="IEtlPipeline{T}"/> for chaining.</returns>
     /// <exception cref="ArgumentNullException"><paramref name="source"/> is <see langword="null"/>.</exception>
-    public static IEtlPipeline<T> From<T>(IAsyncEnumerable<T> source)
+    public IEtlPipeline<T> From<T>(IAsyncEnumerable<T> source)
         where T : notnull
     {
         if (source is null)
@@ -71,7 +85,7 @@ public static IEtlPipeline<T> From<T>(IAsyncEnumerable<T> source)
     /// <param name="extractor">The extractor that seeds the pipeline.</param>
     /// <returns>An <see cref="IEtlPipeline{T}"/> for chaining.</returns>
     /// <exception cref="ArgumentNullException"><paramref name="extractor"/> is <see langword="null"/>.</exception>
-    public static IEtlPipeline<T> From<T, TProgress>(ExtractorBase<T, TProgress> extractor)
+    public IEtlPipeline<T> From<T, TProgress>(ExtractorBase<T, TProgress> extractor)
         where T : notnull
         where TProgress : notnull
     {

src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt

@@ -8,7 +8,6 @@ Wolfgang.Etl.Abstractions.EtlPipelineProgress.RecordsExtracted.get -> int
 Wolfgang.Etl.Abstractions.EtlPipelineProgress.RecordsExtracted.init -> void
 Wolfgang.Etl.Abstractions.EtlPipelineProgress.RecordsLoaded.get -> int
 Wolfgang.Etl.Abstractions.EtlPipelineProgress.RecordsLoaded.init -> void
-Wolfgang.Etl.Abstractions.EtlPipelineSource
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.CurrentItemCount.get -> int
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.CurrentSkippedItemCount.get -> int
@@ -143,9 +142,9 @@ abstract Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.CreatePro
 abstract Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.LoadWorkerAsync(System.Collections.Generic.IAsyncEnumerable<TDestination>! items, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task!
 abstract Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.CreateProgressReport() -> TProgress
 abstract Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.TransformWorkerAsync(System.Collections.Generic.IAsyncEnumerable<TSource>! items, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable<TDestination>!
-static Wolfgang.Etl.Abstractions.EtlPipeline.From<T, TProgress>(Wolfgang.Etl.Abstractions.ExtractorBase<T, TProgress>! extractor) -> Wolfgang.Etl.Abstractions.IEtlPipeline<T>!
-static Wolfgang.Etl.Abstractions.EtlPipeline.From<T>(System.Collections.Generic.IAsyncEnumerable<T>! source) -> Wolfgang.Etl.Abstractions.IEtlPipeline<T>!
-static Wolfgang.Etl.Abstractions.EtlPipeline.Source.get -> Wolfgang.Etl.Abstractions.EtlPipelineSource!
+Wolfgang.Etl.Abstractions.EtlPipeline.From<T, TProgress>(Wolfgang.Etl.Abstractions.ExtractorBase<T, TProgress>! extractor) -> Wolfgang.Etl.Abstractions.IEtlPipeline<T>!
+Wolfgang.Etl.Abstractions.EtlPipeline.From<T>(System.Collections.Generic.IAsyncEnumerable<T>! source) -> Wolfgang.Etl.Abstractions.IEtlPipeline<T>!
+static Wolfgang.Etl.Abstractions.EtlPipeline.Create() -> Wolfgang.Etl.Abstractions.EtlPipeline!
 static Wolfgang.Etl.Abstractions.Pipeline.Extract<TSource, TProgress>(Wolfgang.Etl.Abstractions.IExtractWithProgressAndCancellationAsync<TSource, TProgress>! extractor) -> Wolfgang.Etl.Abstractions.IExtractStageWithProgress<TSource, TProgress>!
 static Wolfgang.Etl.Abstractions.Pipeline.Extract<TSource, TProgress>(Wolfgang.Etl.Abstractions.IExtractWithProgressAsync<TSource, TProgress>! extractor) -> Wolfgang.Etl.Abstractions.IExtractStageWithProgress<TSource, TProgress>!
 static Wolfgang.Etl.Abstractions.Pipeline.Extract<TSource>(Wolfgang.Etl.Abstractions.IExtractAsync<TSource>! extractor) -> Wolfgang.Etl.Abstractions.IExtractStage<TSource>!

tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs

@@ -49,7 +49,9 @@ public async Task RunAsync_when_source_is_IAsyncEnumerable_delivers_all_records_
     {
         var loader = new CollectingLoader<int>();
 
-        await EtlPipeline.From(AsyncSource(1, 2, 3))
+        await EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3))
             .To(loader)
             .RunAsync();
 
@@ -63,7 +65,9 @@ public async Task RunAsync_when_source_is_an_extractor_delivers_all_records_to_t
         var loader = new CollectingLoader<int>();
         var extractor = new SeededExtractor<int>(new[] { 10, 20, 30 });
 
-        await EtlPipeline.From(extractor)
+        await EtlPipeline
+            .Create()
+            .From(extractor)
             .To(loader)
             .RunAsync();
 
@@ -76,7 +80,9 @@ public async Task Through_pipes_records_through_the_transformer()
     {
         var loader = new CollectingLoader<string>();
 
-        await EtlPipeline.From(AsyncSource(1, 2, 3))
+        await EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3))
             .Through(new MapTransformer<int, string>(x => $"n{x}"))
             .To(loader)
             .RunAsync();
@@ -90,7 +96,9 @@ public async Task Through_can_be_chained()
     {
         var loader = new CollectingLoader<int>();
 
-        await EtlPipeline.From(AsyncSource(1, 2, 3))
+        await EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3))
             .Through(new MapTransformer<int, int>(x => x + 1))
             .Through(new MapTransformer<int, int>(x => x * 10))
             .To(loader)
@@ -107,7 +115,9 @@ public async Task Through_forwards_the_cancellation_token_to_a_cancellation_awar
         var transformer = new TokenRecordingTransformer<int>();
         var loader = new CollectingLoader<int>();
 
-        await EtlPipeline.From(AsyncSource(1, 2, 3))
+        await EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3))
             .Through(transformer)
             .To(loader)
             .RunAsync(null, cts.Token);
@@ -122,7 +132,9 @@ public async Task Through_delegate_pipes_records_through_the_stage()
     {
         var loader = new CollectingLoader<string>();
 
-        await EtlPipeline.From(AsyncSource(1, 2, 3))
+        await EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3))
             .Through(Label)                       // Func<IAsyncEnumerable<int>, IAsyncEnumerable<string>>
             .To(loader)
             .RunAsync();
@@ -145,7 +157,9 @@ public async Task Through_cancellation_aware_delegate_forwards_the_token()
             return items;
         };
 
-        await EtlPipeline.From(AsyncSource(1, 2, 3))
+        await EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3))
             .Through(stage)
             .To(loader)
             .RunAsync(null, cts.Token);
@@ -158,7 +172,9 @@ await EtlPipeline.From(AsyncSource(1, 2, 3))
     [Fact]
     public void Through_when_delegate_is_null_throws_ArgumentNullException()
     {
-        var pipeline = EtlPipeline.From(AsyncSource(1));
+        var pipeline = EtlPipeline
+            .Create()
+            .From(AsyncSource(1));
         Assert.Throws<ArgumentNullException>(
             () => pipeline.Through((Func<IAsyncEnumerable<int>, IAsyncEnumerable<int>>)null!));
     }
@@ -167,7 +183,9 @@ public void Through_when_delegate_is_null_throws_ArgumentNullException()
     [Fact]
     public void Through_when_cancellation_aware_delegate_is_null_throws_ArgumentNullException()
     {
-        var pipeline = EtlPipeline.From(AsyncSource(1));
+        var pipeline = EtlPipeline
+            .Create()
+            .From(AsyncSource(1));
         Assert.Throws<ArgumentNullException>(
             () => pipeline.Through((Func<IAsyncEnumerable<int>, CancellationToken, IAsyncEnumerable<int>>)null!));
     }
@@ -176,7 +194,9 @@ public void Through_when_cancellation_aware_delegate_is_null_throws_ArgumentNull
     [Fact]
     public async Task AsAsyncEnumerable_exposes_the_composed_stream()
     {
-        var stream = EtlPipeline.From(AsyncSource(1, 2, 3))
+        var stream = EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3))
             .Through(new MapTransformer<int, int>(x => x * 100))
             .AsAsyncEnumerable();
 
@@ -191,7 +211,9 @@ public async Task RunAsync_propagates_an_exception_thrown_by_a_transformer()
     {
         var loader = new CollectingLoader<int>();
 
-        var sink = EtlPipeline.From(AsyncSource(1, 2, 3))
+        var sink = EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3))
             .Through(new MapTransformer<int, int>(_ => throw new InvalidOperationException("boom")))
             .To(loader);
 
@@ -206,7 +228,9 @@ public async Task RunAsync_observes_cancellation_mid_stream()
         using var cts = new CancellationTokenSource();
         var loader = new CollectingLoader<int>();
 
-        var sink = EtlPipeline.From(AsyncSource(1, 2, 3, 4, 5))
+        var sink = EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3, 4, 5))
             .Through(new CancelingTransformer<int>(cts, 2))
             .To(loader);
 
@@ -222,7 +246,9 @@ public async Task RunAsync_reports_extracted_and_loaded_counters()
         var progress = new SynchronousProgress<EtlPipelineProgress>(reports.Add);
         var loader = new CollectingLoader<int>();
 
-        await EtlPipeline.From(AsyncSource(1, 2, 3, 4, 5))
+        await EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3, 4, 5))
             .To(loader)
             .RunAsync(progress);
 
@@ -233,46 +259,60 @@ await EtlPipeline.From(AsyncSource(1, 2, 3, 4, 5))
 
 
     [Fact]
-    public void Source_sentinel_is_available_for_format_package_extensions()
+    public void Create_returns_a_new_instance_each_call()
     {
-        Assert.NotNull(EtlPipeline.Source);
+        var a = EtlPipeline.Create();
+        var b = EtlPipeline.Create();
+
+        Assert.NotNull(a);
+        Assert.NotSame(a, b);
     }
 
 
     [Fact]
     public void From_when_stream_is_null_throws_ArgumentNullException()
     {
-        Assert.Throws<ArgumentNullException>(() => EtlPipeline.From<int>((IAsyncEnumerable<int>)null!));
+        Assert.Throws<ArgumentNullException>(() => EtlPipeline
+            .Create()
+            .From<int>((IAsyncEnumerable<int>)null!));
     }
 
 
     [Fact]
     public void From_when_extractor_is_null_throws_ArgumentNullException()
     {
-        Assert.Throws<ArgumentNullException>(() => EtlPipeline.From<int, EtlProgress>(null!));
+        Assert.Throws<ArgumentNullException>(() => EtlPipeline
+            .Create()
+            .From<int, EtlProgress>(null!));
     }
 
 
     [Fact]
     public void Through_when_transformer_is_null_throws_ArgumentNullException()
     {
-        var pipeline = EtlPipeline.From(AsyncSource(1));
+        var pipeline = EtlPipeline
+            .Create()
+            .From(AsyncSource(1));
         Assert.Throws<ArgumentNullException>(() => pipeline.Through((ITransformAsync<int, int>)null!));
     }
 
 
     [Fact]
     public void Through_when_cancellation_aware_transformer_is_null_throws_ArgumentNullException()
     {
-        var pipeline = EtlPipeline.From(AsyncSource(1));
+        var pipeline = EtlPipeline
+            .Create()
+            .From(AsyncSource(1));
         Assert.Throws<ArgumentNullException>(() => pipeline.Through((ITransformWithCancellationAsync<int, int>)null!));
     }
 
 
     [Fact]
     public void To_when_loader_is_null_throws_ArgumentNullException()
     {
-        var pipeline = EtlPipeline.From(AsyncSource(1));
+        var pipeline = EtlPipeline
+            .Create()
+            .From(AsyncSource(1));
         Assert.Throws<ArgumentNullException>(() => pipeline.To<EtlProgress>(null!));
     }