dataloader-util (0.3.2)

Published 2026-06-10 19:11:31 +00:00 by gurbakhshish in gurbakhshish/dataloader_util

Installation

pip install --index-url  dataloader-util

About this package

Config-driven CLI for loading tabular data from sources (local, S3) in formats (CSV, JSONL, Parquet) into targets (Postgres, Trino/Iceberg, Snowflake).

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:

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:

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

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

The package exposes the console command:

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:

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:

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:

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

Validate without running:

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:

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

Enable debug logging:

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

CLI reference

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

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

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

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:

format: csv
format_options:
  delimiter: "|"
  header: true
  null_values: ["", "NULL"]
  compression: gzip
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.

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.

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

Rename transform

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

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

Cast transform

Types are DuckDB SQL type names.

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

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

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

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

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.

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:

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:

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:

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

Requirements

Requires Python: >=3.14
Details
PyPI
2026-06-10 19:11:31 +00:00
2
197 KiB
Assets (2)
Versions (10) View all
0.3.8 2026-07-28
0.3.7 2026-07-24
0.3.6 2026-07-16
0.3.5 2026-06-26
0.3.4 2026-06-26