Metadata-Version: 2.4
Name: dataloader-util
Version: 0.1.1
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
```

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

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

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

## Minimal job config

```yaml
name: local_csv_to_local_parquet

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

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

target:
  kind: local
  path: ./output/events.parquet
  format: parquet
  mode: replace
```

## Configuration reference

Top-level fields:

| Field | Required | Description |
|-------|----------|-------------|
| `name` | No | Optional job name shown in logs and validation output. |
| `source` | Yes | Input location and format. |
| `transforms` | No | Ordered list of transforms. Defaults to no transforms. |
| `target` | Yes | Output location and write mode. |
| `connection` | For DB targets | Connection details for Postgres, Trino, or Snowflake. Omit for local targets. |

Unknown keys are rejected at every level.

### Sources

#### Local source

```yaml
source:
  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
source:
  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

Transforms run in the order listed. 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:
  - 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:
  - kind: filter
    where: "amount > 0 AND status = 'active'"
```

#### Rename transform

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

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

#### Cast transform

Types are DuckDB SQL type names.

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

### Targets

#### Local file target

```yaml
target:
  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
target:
  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
target:
  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
target:
  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
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. |

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