diff --git a/CHANGELOG.md b/CHANGELOG.md index 9930e996..0c41c2b0 100644 --- a/CHANGELOG.md +++ b/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)` and `From(ExtractorBase)` factories, - plus the `EtlPipeline.Source` sentinel that format packages extend with source factories. + - `EtlPipeline.Create()` returns a fresh builder seed; `From(IAsyncEnumerable)` and + `From(ExtractorBase)` factories start the chain from any source. Format + packages extend the `EtlPipeline` instance with class-named source factories, e.g. + `EtlPipeline.Create().CsvExtractor("orders.csv")`. - `IEtlPipeline` with `Through` (four overloads — an `ITransformAsync` or `ITransformWithCancellationAsync` transformer, or a stream-to-stream delegate, with or without a `CancellationToken`), `To(LoaderBase)`, and diff --git a/examples/Net4.8/Example8-EtlPipeline/Program.cs b/examples/Net4.8/Example8-EtlPipeline/Program.cs index 716ae1f3..7cd851c7 100644 --- a/examples/Net4.8/Example8-EtlPipeline/Program.cs +++ b/examples/Net4.8/Example8-EtlPipeline/Program.cs @@ -27,7 +27,9 @@ private static async Task Main() // Each Through returns IEtlPipeline, 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 is a valid source via EtlPipeline.From(...). - // An ExtractorBase works too: EtlPipeline.From(myExtractor). + // Any IAsyncEnumerable is a valid source via EtlPipeline.Create().From(...). + // An ExtractorBase works too: EtlPipeline.Create().From(myExtractor). private static async IAsyncEnumerable RawNumbers() { for (var i = 1; i <= 8; i++) diff --git a/examples/Net8.0/Example8-EtlPipeline/Program.cs b/examples/Net8.0/Example8-EtlPipeline/Program.cs index 19b320da..4d46e3b4 100644 --- a/examples/Net8.0/Example8-EtlPipeline/Program.cs +++ b/examples/Net8.0/Example8-EtlPipeline/Program.cs @@ -24,7 +24,9 @@ private static async Task Main() // Each Through returns IEtlPipeline, 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 is a valid source via EtlPipeline.From(...). - // An ExtractorBase works too: EtlPipeline.From(myExtractor). + // Any IAsyncEnumerable is a valid source via EtlPipeline.Create().From(...). + // An ExtractorBase works too: EtlPipeline.Create().From(myExtractor). private static async IAsyncEnumerable RawNumbers() { for (var i = 1; i <= 8; i++) diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipeline.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipeline.cs index d2d1a465..444c21fe 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipeline.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipeline.cs @@ -6,10 +6,11 @@ namespace Wolfgang.Etl.Abstractions; /// -/// Entry point for building a generic, format-agnostic ETL pipeline. Start from a source — either a -/// built-in / -/// factory, or a format-specific factory hung off by a format package — chain -/// append transformer stages on , terminate with a sink, then call +/// Entry point for building a generic, format-agnostic ETL pipeline. Obtain a fresh builder with +/// , start from a source — either a built-in / +/// factory, or a format-specific factory +/// hung off the instance by a format package — chain append transformer stages +/// on , terminate with a sink, then call /// . /// /// @@ -19,6 +20,13 @@ namespace Wolfgang.Etl.Abstractions; /// System.IO.Pipelines. /// /// +/// returns a fresh 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 +/// public static ICsvExtractorBuilder<T> CsvExtractor<T>(this EtlPipeline pipeline, string path), +/// enabling EtlPipeline.Create().CsvExtractor<Order>("orders.csv"). +/// +/// /// The core exposes only the plumbing — a source, /// for appending transformer stages, and a sink. The LINQ-flavored operators (Where, /// Select, Distinct, …) are extension methods shipped by Wolfgang.Etl.Transformers, @@ -27,19 +35,25 @@ namespace Wolfgang.Etl.Abstractions; /// /// /// -/// await EtlPipeline.From(records) +/// await EtlPipeline.Create() +/// .From(records) /// .Through(new WhereTransformer<Order>(r => r.Amount > 0)) /// .To(sqlLoader) /// .RunAsync(progress, token); /// /// -public static class EtlPipeline +public sealed class EtlPipeline { + private EtlPipeline() + { + } + + /// - /// The sentinel that format packages extend with source factories, enabling the - /// EtlPipeline.Source.CsvExtractor<T>(...) shape. See . + /// Creates a new, empty builder to start a fluent pipeline chain. /// - public static EtlPipelineSource Source { get; } = new(); + /// A fresh builder instance. + public static EtlPipeline Create() => new(); /// @@ -50,7 +64,7 @@ public static class EtlPipeline /// The stream that seeds the pipeline. /// An for chaining. /// is . - public static IEtlPipeline From(IAsyncEnumerable source) + public IEtlPipeline From(IAsyncEnumerable source) where T : notnull { if (source is null) @@ -71,7 +85,7 @@ public static IEtlPipeline From(IAsyncEnumerable source) /// The extractor that seeds the pipeline. /// An for chaining. /// is . - public static IEtlPipeline From(ExtractorBase extractor) + public IEtlPipeline From(ExtractorBase extractor) where T : notnull where TProgress : notnull { diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSource.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSource.cs deleted file mode 100644 index 193bb589..00000000 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSource.cs +++ /dev/null @@ -1,15 +0,0 @@ -namespace Wolfgang.Etl.Abstractions; - -/// -/// A sentinel target for source-factory extension methods. C# cannot attach extension methods to a -/// static class, so exposes an instance of this type for format -/// packages to extend — for example a CSV package adds -/// public static ICsvExtractorBuilder<T> CsvExtractor<T>(this EtlPipelineSource source, string path), -/// enabling EtlPipeline.Source.CsvExtractor<Order>("orders.csv"). -/// -public sealed class EtlPipelineSource -{ - internal EtlPipelineSource() - { - } -} diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt index a4869739..71f7efaa 100644 --- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt +++ b/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 Wolfgang.Etl.Abstractions.ExtractorBase.CurrentItemCount.get -> int Wolfgang.Etl.Abstractions.ExtractorBase.CurrentSkippedItemCount.get -> int @@ -143,9 +142,9 @@ abstract Wolfgang.Etl.Abstractions.LoaderBase.CreatePro abstract Wolfgang.Etl.Abstractions.LoaderBase.LoadWorkerAsync(System.Collections.Generic.IAsyncEnumerable! items, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task! abstract Wolfgang.Etl.Abstractions.TransformerBase.CreateProgressReport() -> TProgress abstract Wolfgang.Etl.Abstractions.TransformerBase.TransformWorkerAsync(System.Collections.Generic.IAsyncEnumerable! items, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! -static Wolfgang.Etl.Abstractions.EtlPipeline.From(Wolfgang.Etl.Abstractions.ExtractorBase! extractor) -> Wolfgang.Etl.Abstractions.IEtlPipeline! -static Wolfgang.Etl.Abstractions.EtlPipeline.From(System.Collections.Generic.IAsyncEnumerable! source) -> Wolfgang.Etl.Abstractions.IEtlPipeline! -static Wolfgang.Etl.Abstractions.EtlPipeline.Source.get -> Wolfgang.Etl.Abstractions.EtlPipelineSource! +Wolfgang.Etl.Abstractions.EtlPipeline.From(Wolfgang.Etl.Abstractions.ExtractorBase! extractor) -> Wolfgang.Etl.Abstractions.IEtlPipeline! +Wolfgang.Etl.Abstractions.EtlPipeline.From(System.Collections.Generic.IAsyncEnumerable! source) -> Wolfgang.Etl.Abstractions.IEtlPipeline! +static Wolfgang.Etl.Abstractions.EtlPipeline.Create() -> Wolfgang.Etl.Abstractions.EtlPipeline! static Wolfgang.Etl.Abstractions.Pipeline.Extract(Wolfgang.Etl.Abstractions.IExtractWithProgressAndCancellationAsync! extractor) -> Wolfgang.Etl.Abstractions.IExtractStageWithProgress! static Wolfgang.Etl.Abstractions.Pipeline.Extract(Wolfgang.Etl.Abstractions.IExtractWithProgressAsync! extractor) -> Wolfgang.Etl.Abstractions.IExtractStageWithProgress! static Wolfgang.Etl.Abstractions.Pipeline.Extract(Wolfgang.Etl.Abstractions.IExtractAsync! extractor) -> Wolfgang.Etl.Abstractions.IExtractStage! diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs index f2d436b2..5b3a823c 100644 --- a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs +++ b/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(); - 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(); var extractor = new SeededExtractor(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(); - await EtlPipeline.From(AsyncSource(1, 2, 3)) + await EtlPipeline + .Create() + .From(AsyncSource(1, 2, 3)) .Through(new MapTransformer(x => $"n{x}")) .To(loader) .RunAsync(); @@ -90,7 +96,9 @@ public async Task Through_can_be_chained() { var loader = new CollectingLoader(); - await EtlPipeline.From(AsyncSource(1, 2, 3)) + await EtlPipeline + .Create() + .From(AsyncSource(1, 2, 3)) .Through(new MapTransformer(x => x + 1)) .Through(new MapTransformer(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(); var loader = new CollectingLoader(); - 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(); - await EtlPipeline.From(AsyncSource(1, 2, 3)) + await EtlPipeline + .Create() + .From(AsyncSource(1, 2, 3)) .Through(Label) // Func, IAsyncEnumerable> .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( () => pipeline.Through((Func, IAsyncEnumerable>)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( () => pipeline.Through((Func, CancellationToken, IAsyncEnumerable>)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(x => x * 100)) .AsAsyncEnumerable(); @@ -191,7 +211,9 @@ public async Task RunAsync_propagates_an_exception_thrown_by_a_transformer() { var loader = new CollectingLoader(); - var sink = EtlPipeline.From(AsyncSource(1, 2, 3)) + var sink = EtlPipeline + .Create() + .From(AsyncSource(1, 2, 3)) .Through(new MapTransformer(_ => 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(); - var sink = EtlPipeline.From(AsyncSource(1, 2, 3, 4, 5)) + var sink = EtlPipeline + .Create() + .From(AsyncSource(1, 2, 3, 4, 5)) .Through(new CancelingTransformer(cts, 2)) .To(loader); @@ -222,7 +246,9 @@ public async Task RunAsync_reports_extracted_and_loaded_counters() var progress = new SynchronousProgress(reports.Add); var loader = new CollectingLoader(); - await EtlPipeline.From(AsyncSource(1, 2, 3, 4, 5)) + await EtlPipeline + .Create() + .From(AsyncSource(1, 2, 3, 4, 5)) .To(loader) .RunAsync(progress); @@ -233,30 +259,40 @@ 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(() => EtlPipeline.From((IAsyncEnumerable)null!)); + Assert.Throws(() => EtlPipeline + .Create() + .From((IAsyncEnumerable)null!)); } [Fact] public void From_when_extractor_is_null_throws_ArgumentNullException() { - Assert.Throws(() => EtlPipeline.From(null!)); + Assert.Throws(() => EtlPipeline + .Create() + .From(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(() => pipeline.Through((ITransformAsync)null!)); } @@ -264,7 +300,9 @@ public void Through_when_transformer_is_null_throws_ArgumentNullException() [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(() => pipeline.Through((ITransformWithCancellationAsync)null!)); } @@ -272,7 +310,9 @@ public void Through_when_cancellation_aware_transformer_is_null_throws_ArgumentN [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(() => pipeline.To(null!)); }