dataloader-util (0.1.1)

Published 2026-06-06 04:50:57 +00:00 by gurbakhshish in gurbakhshish/dataloader_util

To install the package using pip, run the following command:

pip install --index-url  dataloader-util

For more information on the PyPI registry, see the documentation.

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

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

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

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

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

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:

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

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.

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.

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

Rename transform

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

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

Cast transform

Types are DuckDB SQL type names.

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

Targets

Local file target

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

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

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

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.

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:

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

Requires Python: >=3.14

Details

PyPI

gurbakhshish/dataloader_util

2026-06-06 04:50:57 +00:00

185 KiB

Assets (2)

dataloader_util-0.1.1.tar.gz 144 KiB

dataloader_util-0.1.1-py3-none-any.whl 41 KiB

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

Issues

dataloader-util (0.1.1)

Installation

About this package

dataloader-util

What it does

Requirements

Installation and setup

Quickstart: local CSV to local Parquet

CLI reference

Minimal job config

Configuration reference

Sources

Local source

S3 source

Formats

Transforms

SQL transform

Filter transform

Rename transform

Cast transform

Targets

Local file target

Postgres target

Trino target

Snowflake target

Write modes

Environment variable substitution

Included examples

Local integration services

Development

Troubleshooting

License

Requirements