dtpipe 1.6.0

DtPipe

A self-contained CLI for streaming, transforming, and anonymizing data across databases and file formats.

DtPipe reads from a source, applies row and columnar transformations in batches, and writes to a destination with no intermediate staging. It is designed for automation and CI/CD workflows where repeatable, observable data pipelines matter.

📖 Recipes & Examples → COOKBOOK.md · Full CLI Reference → REFERENCE.md

Installation

.NET Global Tool (Recommended)

dotnet tool install -g dtpipe
dtpipe --help

Build from Source

Prerequisite: .NET 10 SDK

# Bash (Mac/Linux/Windows Git Bash)
./build.sh

# PowerShell (Windows/Cross-platform)
./build.ps1

Binary created at: ./dist/release/dtpipe

Quick Start

Export a database table

dtpipe \
  -i "pg:Host=localhost;Database=prod;Username=postgres" \
  --query "SELECT * FROM users" \
  -o users.parquet

Anonymize before export

dtpipe \
  -i "pg:Host=localhost;Database=prod;Username=postgres" \
  --query "SELECT * FROM users" \
  --fake "email:internet.email" \
  --fake "name:name.fullName" \
  --mask "phone:###-****" \
  --null "ssn" \
  -o anonymized_users.parquet

In-memory SQL join

dtpipe \
  -i orders.parquet --alias orders \
  -i customers.csv --alias customers \
  --from orders --ref customers \
  --sql "SELECT o.*, c.name FROM orders o JOIN customers c ON o.customer_id = c.id" \
  -o result.parquet

Run from a YAML job file

# Generate a reusable job file from any CLI command
dtpipe -i "pg:..." --query "SELECT * FROM users" --fake "email:internet.email" \
       -o users.parquet --export-job nightly.yaml

# Run it (with optional overrides)
dtpipe --job nightly.yaml --limit 1000

Incremental loading (cursor-driven)

# Full load on first run (state file does not exist → uses default 1970-01-01)
dtpipe \
  -i "pg:Host=localhost;Database=prod" \
  --query "SELECT * FROM users WHERE updated_at >= '${{cursor://state.json|1970-01-01}}'" \
  -o "sqlite:Data Source=dw.db" --table "users" --strategy Recreate --key id \
  --cursor "updated_at" --state "state.json"

# Subsequent runs: cursor is resolved from state.json, only newer rows are fetched (switch to Upsert)
# See REFERENCE.md#incremental-loading for full flag table and state file format,
# and COOKBOOK.md#incremental-loading for the complete recipe.

Database Resilience (Retry Policy)

# Automatically retry transient database connection and timeout errors with Polly exponential backoff
dtpipe \
  -i "pg:Host=localhost;Database=prod" \
  --query "SELECT * FROM sales" \
  -o "mssql:Server=remote_db;Database=analytics" \
  --retry

Start AI Agent MCP Server (Model Context Protocol)

# Launch native MCP server over STDIO for AI assistants (Cursor, Claude Desktop, Antigravity)
dtpipe mcp

Includes tools: dry-run, suggest-pipeline, list-cursors, execute-yaml-job, and schema discovery.

Interactive AI Agent Mode

# Launch the interactive AI agent (supports Ollama & OpenAI backends)
dtpipe agent --provider openai --api-key "sk-..."

# Or run a one-shot mission
dtpipe agent "Inspect csv:invoices.csv, anonymize email, and output to jsonl:users.jsonl"

Features local Ollama auto-discovery, official OpenAI SDK integration, Spectre.Console TUI, step-by-step trajectory inspector, Spectre DAG topology rendering, and 1-click YAML pipeline export.

Providers

DtPipe detects providers from file extensions (.csv, .parquet…) or explicit prefixes — explicit prefixes are recommended to avoid ambiguity. Full table with capabilities, query requirements and Stdin/Stdout support: REFERENCE.md#providers.

Use keyring://alias anywhere a connection string is expected. DtPipe resolves it from the OS keychain at runtime. Run dtpipe secret set prod-db "pg:..." to store a secret.

For object storage (S3, GCS, Azure Blob), Iceberg, MySQL/MariaDB, HTTP APIs, spatial formats — use DuckDB's extension ecosystem as a connector multiplier. Load an extension with --duck-init on any DuckDB reader, writer, or --sql branch — no additional adapter required. See REFERENCE.md#provider-specific-options and COOKBOOK.md#duckdb-extensions-and-cloud-storage.

Key Concepts

• Providers — where data comes from and goes to. DtPipe reads from databases (pg:, mssql:, ora:, sqlite:, duck:) and files (csv:, parquet:, jsonl:…), and writes to the same set. The provider is inferred from the file extension or an explicit prefix. See Providers for the full list.
• Transformers — what happens in between. Flags like --fake, --mask, --compute, --filter, --rename are chained left-to-right on every row. Example: anonymize, then derive a column, then filter: --fake "email:internet.email" --compute "fullName:row.first+' '+row.last" --filter "row.age>=18". See COOKBOOK.md.
• DAG pipelines — combine or split streams. Use --alias to name a source, then --from / --ref / --sql / --merge to join, union or fan-out without temp files. Typical uses: enrich a stream with a lookup table, or write one source to two sinks at once. See DAG Syntax and DAG recipes.
• YAML jobs — make it repeatable. Any CLI pipeline can be saved with --export-job pipeline.yaml and replayed with dtpipe --job pipeline.yaml (CI/CD, cron, overrides via CLI). See YAML Job Schema.
• Secrets — keep credentials out of shell history. dtpipe secret set prod-db "pg:..." stores in the OS keychain; reference as keyring://prod-db or ${{keyring://alias}} anywhere a connection string is expected. See Secret Management.

Why DuckDB? DuckDB is fast, self-contained and speaks rich SQL — DtPipe embeds it as its SQL engine (--sql, duck:). When DuckDB alone covers your use case, use it directly. DtPipe adds value where DuckDB stops: anonymization/masking in transit, concurrent fan-out, database write strategies (upsert, auto-migrate, bulk), Oracle/SQL Server/XML sources, and repeatable YAML jobs with secret management.

Documentation

Shell Autocompletion (experimental)

dtpipe completion --install

Restart your terminal (or source ~/.zshrc) to activate.

Contributing

See EXTENDING.md for the adapter and transformer patterns.

License

MIT