Kanject.Core.EtlTaskManager

View .md

A small, in-process Extract → Transform → Load framework for bounded import and migration work. Derive a task from AbstractEtlTaskDefinitionManager<TExtract, TTransform>, point it at a data extractor, and the base runs the pipeline: extract rows, hand them to your Transform, then your LoadAsync. Built-in extractors cover CSV, JSON, DynamoDB, and app secrets.

Install

bash 
dotnet add package Kanject.Core.EtlTaskManager

Define a task

csharp 
using Kanject.Core.EtlTaskManager.Abstractions;using Kanject.Core.EtlTaskManager.DataProviders.CsvDataProvider.Extractors;​// Derive from AbstractEtlTaskDefinitionManager<TExtract, TTransform>.// Set an extractor provider in the ctor, override Transform, implement LoadAsync.public sealed class ProductCatalogImportTask    : AbstractEtlTaskDefinitionManager<ProductImportRow, CatalogProduct>{    private readonly IProductCatalog _catalog;​    public ProductCatalogImportTask(ProductImportOptions options, IProductCatalog catalog)    {        _catalog = catalog;        DataExtractorProvider = new CsvDataExtractorProvider(options.SourceFile);    }​    public override void Transform(IEnumerable<ProductImportRow> rows)        => TransformedData = rows            .Where(r => !string.IsNullOrWhiteSpace(r.Sku))            .Select(r => new CatalogProduct { Sku = r.Sku.Trim(), Name = r.Name.Trim() })            .ToArray();​    public override Task LoadAsync() => _catalog.UpsertAsync(TransformedData);}

The base handles the flow. You supply three things: a DataExtractorProvider (where rows come from), a Transform (shape TExtract into TTransform, stored on TransformedData), and a LoadAsync (write the transformed set out). Built-in providers include CsvDataExtractorProvider, JsonDataTransformerProvider, DynamoDbExtractorProvider, and AppSecretsExtractorProvider.

Register

csharp 
using Kanject.Core.EtlTaskManager.Abstractions.Extensions;​// Register each task with the container.services.AddEtlPackage<ProductCatalogImportTask>();

Run it

Run tasks explicitly by resolving them in a DI scope:

csharp 
// Run explicitly: resolve the task(s) in a scope and await RunETLAsync().await using var scope = serviceProvider.CreateAsyncScope();​var tasks = scope.ServiceProvider.GetServices<IEtlTaskDefinition>();await Task.WhenAll(tasks.Select(t => t.RunETLAsync()));

Or run every registered task once at application startup:

csharp 
// …or run every registered task at ASP.NET Core startup.app.UseEtlPackageManager(options => options.RunEtlPackageManager = true);

What ships with it

  • AbstractEtlTaskDefinitionManager<TExtract, TTransform> — override Transform, implement LoadAsync, set DataExtractorProvider
  • Single-type AbstractEtlTaskDefinitionManager<TModel> for extract-and-load without a separate transform type
  • Built-in extractors: CSV, JSON, DynamoDB, and app secrets
  • AddEtlPackage<T>() registration; RunETLAsync() to execute, or UseEtlPackageManager(...) to run all at startup
  • Startup runner executes registered tasks concurrently with Task.WhenAll
Was this page helpful?