Metadata-Version: 2.4
Name: dataloader-util
Version: 0.3.2
Summary: Config-driven CLI for loading tabular data from sources (local, S3) in formats (CSV, JSONL, Parquet) into targets (Postgres, Trino/Iceberg, Snowflake).
Requires-Python: >=3.14
Requires-Dist: boto3>=1.43.20
Requires-Dist: duckdb>=1.5.3
Requires-Dist: loguru>=0.7.3
Requires-Dist: psycopg[binary]>=3.3.4
Requires-Dist: pyarrow>=24.0.0
Requires-Dist: pydantic>=2.13.4
Requires-Dist: pyyaml>=6.0.3
Requires-Dist: snowflake-connector-python[pandas]>=4.6.0
Requires-Dist: sqlalchemy>=2
Requires-Dist: trino>=0.337.0
Requires-Dist: typer>=0.26.6
Description-Content-Type: text/markdown

# dataloader-util

`dataloader-util` is a config-driven Python CLI for moving tabular data from files or S3 into local files and analytical databases. It uses DuckDB as the in-process execution engine, so jobs can read common data formats, run SQL-style transformations, infer schemas, and write to targets from a single YAML or JSON config file.

## What it does

A load job follows one pipeline:

```text
source file(s) -> DuckDB view -> optional transforms -> target writer
```

A config file declares **named pools** of sources, targets, and
transform groups. The CLI picks one of each by name at run time:

```text
sources:   { s3_csv: ..., local_parquet: ... }
targets:   { warehouse: ..., archive: ... }
transforms: { clean: [...steps], camel: [...steps] }

# CLI:
dataloader-util load -c job.yaml \
    --source s3_csv --target warehouse \
    --transform clean --transform camel
```

This lets a single config file describe many pipelines (different
sources, destinations, or transform chains) and pick one per run
without rewriting the file.

Supported capabilities:

- **Sources**: local filesystem paths/globs and S3 or S3-compatible object stores.
- **Input formats**: CSV, JSONL/NDJSON, and Parquet.
- **Targets**: local Parquet/CSV files, Postgres, Trino, Trino-managed Iceberg tables, and Snowflake.
- **Transforms**: free-form DuckDB SQL, `WHERE` filters, column casts, and column renames.
- **Write modes**: `replace`, `append`, and key-based `merge` for database targets.
- **Config validation**: strict Pydantic validation catches unknown keys and invalid combinations before a job runs.
- **Secrets via environment**: `${VAR}` placeholders are resolved from environment variables at runtime.

## Requirements

- Python `>=3.14`
- [`uv`](https://docs.astral.sh/uv/) for local development commands
- Docker only if you want to run local integration services/tests
- Target-specific connectivity for Postgres, Trino, Snowflake, or S3

## Installation and setup

From a checkout of this repository:

```bash
uv sync
uv run dataloader-util --help
uv run dataloader-util --version
```

The package exposes the console command:

```bash
dataloader-util
```

During development, prefer running it through `uv run` so the repository virtual environment is used.

## Quickstart: local CSV to local Parquet

Create a small input file:

```bash
mkdir -p data output
cat > data/events.csv <<'CSV'
id,event_ts,user_id,amount
1,2026-01-01T00:00:00Z,u1,12.50
2,2026-01-02T00:00:00Z,u2,-3.00
3,2026-01-03T00:00:00Z,u3,7.25
CSV
```

Run the included example job:

```bash
uv run dataloader-util load --config examples/job_local_to_local.yaml
```

With one source and one target defined, the CLI picks them
implicitly. To pick a specific source/target by name, use
`--source` / `--target`:

```bash
uv run dataloader-util load --config examples/job_multi_source_target.yaml \
    --source events_csv --target events_parquet \
    --transform keep_positive
```

Validate without running:

```bash
uv run dataloader-util validate --config examples/job_local_to_local.yaml
```

Perform a dry run that reads the source, applies transforms, and prints the planned target information without writing:

```bash
uv run dataloader-util load --dry-run --config examples/job_local_to_local.yaml
```

Enable debug logging:

```bash
uv run dataloader-util load --log-level DEBUG --config examples/job_local_to_local.yaml
```

## CLI reference

```bash
dataloader-util [OPTIONS] COMMAND [ARGS]...
```

Global options:

| Option | Description |
|--------|-------------|
| `--version` | Print the installed version and exit. |
| `--help` | Show CLI help. |

Commands:

| Command | Description |
|---------|-------------|
| `load --config/-c PATH [--source NAME] [--target NAME] [--transform NAME]...` | Validate and execute a load job. |
| `load --dry-run --config/-c PATH [...]` | Validate, read the source, apply transforms, and report target planning without writing. |
| `load --log-level/-L LEVEL --config/-c PATH [...]` | Run with a Loguru level such as `DEBUG`, `INFO`, or `ERROR`. |
| `validate --config/-c PATH [...]` | Validate config shape and print a summary without reading or writing data. |

`--source` and `--target` are optional when the config defines
exactly one of each; required (and an unknown name is rejected with
a list of available names) when multiple are defined.
`--transform` is repeatable and optional: each occurrence names a
group from `transforms:` to concatenate and apply in order. Omit
`--transform` entirely to run with no transforms.

Config files can be YAML (`.yaml`, `.yml`) or JSON (`.json`).

## Minimal job config

```yaml
name: local_csv_to_local_parquet

sources:
  events_csv:
    kind: local
    path: ./data/events.csv
    format: csv
    format_options:
      delimiter: ","
      header: true
      null_values: ["", "null"]

targets:
  events_parquet:
    kind: local
    path: ./output/events.parquet
    format: parquet
    mode: replace

transforms:
  keep_positive:
    - kind: sql
      sql: "SELECT id, event_ts, user_id, amount FROM <source> WHERE amount > 0"
```

## Configuration reference

Top-level fields:

| Field | Required | Description |
|-------|----------|-------------|
| `name` | No | Optional pipeline name shown in logs and validation output. |
| `sources` | No | Map of named source configs. At least one is needed to run. |
| `targets` | No | Map of named target configs. At least one is needed to run. |
| `transforms` | No | Map of named transform groups; each value is an ordered list of transform steps. |

When the config has a single source, a single target, and no
transforms, the CLI runs it with no flags. With more than one of
any kind, pass `--source`, `--target`, and/or `--transform` to
select.

Each named target embeds its own `connection:` block (required for
`postgres`/`trino`/`snowflake`, omitted for `local`). Unknown keys
are rejected at every level.

### Sources

#### Local source

```yaml
sources:
  events:
    kind: local
    path: ./data/events/*.csv
    format: csv
    format_options:
      header: true
```

- `path` can be a single file or a DuckDB-supported glob.
- Non-glob local paths are checked before reading.

#### S3 source

```yaml
sources:
  events:
    kind: s3
    path: s3://my-bucket/raw/events/*.parquet
    format: parquet
    format_options:
      hive_partitioning: true
    region: us-east-1
    endpoint_url: http://localhost:9000   # optional, for MinIO/LocalStack
    access_key_id: ${AWS_ACCESS_KEY_ID}   # optional
    secret_access_key: ${AWS_SECRET_ACCESS_KEY} # optional
```

- S3 reads use DuckDB's `httpfs` extension.
- If explicit keys are omitted, DuckDB/AWS credential discovery is used.
- Set `endpoint_url` for S3-compatible systems such as MinIO or LocalStack.

### Formats

| Format | Reader | Options |
|--------|--------|---------|
| `csv` | DuckDB `read_csv_auto` | `delimiter`, `header`, `null_values`, `compression` (`none`, `gzip`, `zstd`) |
| `jsonl` | DuckDB `read_json_auto` with newline-delimited format | `compression` (`none`, `gzip`, `zstd`) |
| `parquet` | DuckDB `read_parquet` | `columns`, `hive_partitioning` |

Examples:

```yaml
format: csv
format_options:
  delimiter: "|"
  header: true
  null_values: ["", "NULL"]
  compression: gzip
```

```yaml
format: parquet
format_options:
  columns: [id, event_ts, amount]
  hive_partitioning: true
```

### Transforms

A `transforms:` entry maps a group name to an ordered list of
transform steps. The CLI concatenates the steps from each
`--transform NAME` flag in the order the flags are given. Each
transform reads the previous DuckDB view and registers a new
temporary view.

#### SQL transform

Use `<source>` as the placeholder for the current input view.

```yaml
transforms:
  keep_positive:
    - kind: sql
      sql: "SELECT id, amount * 100 AS amount_cents FROM <source> WHERE amount > 0"
```

#### Filter transform

The `where` value is the expression only; do not include `WHERE`.

```yaml
transforms:
  active_only:
    - kind: filter
      where: "amount > 0 AND status = 'active'"
```

#### Rename transform

Mapping is `old_name: new_name`. Unmapped columns pass through unchanged.

```yaml
transforms:
  camel_case:
    - kind: rename
      mapping:
        event_ts: eventTs
        user_id: userId
```

#### Cast transform

Types are DuckDB SQL type names.

```yaml
transforms:
  cast_clean:
    - kind: cast
      columns:
        user_id: VARCHAR
        amount: DOUBLE
        event_ts: TIMESTAMP
```

### Targets

Targets live in the top-level `targets:` map, keyed by name. Each
target embeds its own `connection:` block (required for
`postgres`/`trino`/`snowflake`, omitted for `local`).

#### Local file target

```yaml
targets:
  events_parquet:
    kind: local
    path: ./output/events.parquet
    format: parquet
    mode: replace
```

- Supported output formats: `parquet`, `csv`.
- Parent directories are created automatically.
- Use `mode: replace`; local output is written with DuckDB `COPY` to the configured file.

#### Postgres target

```yaml
targets:
  warehouse:
    kind: postgres
    table: analytics.events
    mode: replace
    batch_size: 50000
    connection:
      kind: postgres
      host: localhost
      port: 5432
      user: etl
      password: ${POSTGRES_PASSWORD}
      database: analytics
      schema: public
```

- Table names can be `table` or `schema.table`; a bare table defaults to `public`.
- `replace` drops and recreates the table.
- `append` creates the table if missing and inserts rows.
- `merge` creates the table with a primary key if missing, stages rows in a temp table, then uses `INSERT ... ON CONFLICT`.
- Bulk loading uses `psycopg` `COPY FROM STDIN`.

#### Trino target

```yaml
targets:
  events_iceberg:
    kind: trino
    table: iceberg.raw.events
    mode: append
    batch_size: 5000
    connection:
      kind: trino
      host: localhost
      port: 8080
      user: etl
      catalog: iceberg
      schema: raw
```

- Table names must be `catalog.schema.table`.
- The target schema is created if missing.
- Writes use a temporary table plus server-side insert or merge.
- Iceberg support is provided through Trino's Iceberg connector configuration on the Trino server.

#### Snowflake target

```yaml
targets:
  events_merged:
    kind: snowflake
    table: RAW.EVENTS
    mode: merge
    merge_keys: [id]
    batch_size: 100000
    connection:
      kind: snowflake
      account: ${SNOWFLAKE_ACCOUNT}
      user: ${SNOWFLAKE_USER}
      password: ${SNOWFLAKE_PASSWORD}
      warehouse: COMPUTE_WH
      database: ANALYTICS
      schema: RAW
      role: ${SNOWFLAKE_ROLE}
```

- Table names can be `table` or `schema.table`; the database comes from the connection.
- Writes use `snowflake-connector-python` `write_pandas`.
- `merge` stages data into a temporary table and runs a Snowflake `MERGE` statement.
- Snowflake identifiers are validated as unquoted identifier names.

## Write modes

| Mode | Local | Postgres | Trino | Snowflake |
|------|-------|----------|-------|-----------|
| `replace` | Write/overwrite output file | Drop and recreate table | Drop and recreate table | Drop table, then write with auto-create |
| `append` | Not intended for local files | Create table if needed, then append | Create table if needed, then append | Create table if needed, then append |
| `merge` | Not supported for local files | Upsert by `merge_keys` with `ON CONFLICT` | `MERGE INTO ... USING` | `MERGE INTO ... USING` |

For `merge`, `target.merge_keys` is required and each key must exist in the transformed source columns.

## Environment variable substitution

Any string value can contain `${VAR}`. Placeholders are resolved before schema validation.

```yaml
targets:
  warehouse:
    kind: postgres
    table: analytics.events
    connection:
      kind: postgres
      password: ${POSTGRES_PASSWORD}
```

Rules:

- Missing environment variables raise `ConfigError`.
- Use environment variables for credentials and account-specific values.
- Do not commit real secrets to config files.

## Included examples

| File | Purpose |
|------|---------|
| `examples/job_local_to_local.yaml` | Local CSV to local Parquet with SQL filtering. |
| `examples/job_local_to_local_with_transforms.yaml` | Local CSV to Parquet using filter, rename, and cast transforms. |
| `examples/job_local_to_postgres.yaml` | Local CSV to Postgres replace load. |
| `examples/job_local_to_postgres_merge.yaml` | Local CSV to Postgres upsert by key. |
| `examples/job_s3_to_local.yaml` | S3 Parquet to local Parquet. |
| `examples/job_s3_to_postgres_merge.yaml` | S3 Parquet to Postgres merge. |
| `examples/job_s3_to_snowflake.yaml` | S3 Parquet to Snowflake append. |
| `examples/job_local_to_trino_iceberg.yaml` | Local CSV to an Iceberg table through Trino. |
| `examples/job_multi_source_target.yaml` | One file with multiple sources, targets, and transform groups; pick with `--source` / `--target` / `--transform`. |

## Local integration services

The repository includes `docker-compose.yml` for local integration dependencies:

```bash
docker compose up -d
just test-integration
docker compose down
```

Services include:

- MinIO on `localhost:9000` with console on `localhost:9001`
- Postgres on `localhost:5432` (`etl` / `etlpass` / `analytics`)
- Trino on `localhost:8080` with the memory catalog

Snowflake integration tests are gated by `SNOWFLAKE_*` environment variables and require access to a real account.

## Development

Common tasks are defined in `justfile`:

```bash
just              # list available tasks
just dev          # uv sync
just test         # unit tests only
just test-integration
just test-all
just test-cov
just lint
just fmt          # format check
just fmt-write    # apply formatting
just types
just check        # lint + types + unit tests
just fix          # ruff auto-fix + format
```

Build artifacts can be produced with:

```bash
uv build
```

## Troubleshooting

- **`config file not found`**: check the `--config` path and current working directory.
- **`unsupported config file extension`**: use `.yaml`, `.yml`, or `.json`.
- **Validation errors**: remove unknown keys, ensure target and connection kinds match, and provide `merge_keys` for merge jobs.
- **S3 read failures**: verify region, credentials, bucket path, and `endpoint_url` for S3-compatible services.
- **Trino table errors**: use a fully qualified `catalog.schema.table` target name and verify the catalog exists on the server.
- **Snowflake identifier errors**: use simple unquoted-style database, schema, and table names.

## License

TBD
