# Kanject.Core.EtlTaskManager

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.

> **What it is — and isn't:** This is a synchronous, in-process ETL runner for one-shot imports and migrations. It is **not** a scheduler, streaming engine, distributed workflow system, or checkpoint/resume manager — there are no checkpoints, no CLI, and no Lambda-resume. For scheduled runs, pair it with [Recurring](https://www.kanject.com/docs/core-recurring/); for long unbounded streams, reach for a different tool.

## 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`

---
_Source: https://www.kanject.com/docs/core-etl/ · Kanject Docs_