# Oxy Hq Skills Oxy Etl Builder > ETL Pipeline Builder - Skill: `tomevault-io/oxy-hq-skills-oxy-etl-builder` (Agent Skill, multi-file: 2 files) - Install (CLI): `npx skillmds@latest add tomevault-io/oxy-hq-skills-oxy-etl-builder` - Raw SKILL.md: https://api.skillmd.com/api/skills/tomevault-io/oxy-hq-skills-oxy-etl-builder/raw - Safety review: pending (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - Author: tomevault-io (https://skillmd.com/u/tomevault-io) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/tomevault-io/oxy-hq-skills-oxy-etl-builder --- # ETL Pipeline Builder You are an expert at building ETL (Extract-Transform-Load) pipelines using DLT (data-load-tools). Your role is to help users create robust, maintainable data pipelines that extract from APIs or files and load into data warehouses. ## Scenario Detection Before starting, determine the current state: ### New Project (no `etl/` directory) 1. Set up the core framework first (see Core Setup below) 2. Then proceed to source type classification ### Existing Project (`etl/` directory exists) Skip directly to source type classification - the framework is already in place. ```bash # Check project state ls -la etl/core/pipeline.py 2>/dev/null && echo "Core exists" || echo "New project" ``` ## Source Type Classification After scenario detection, classify what you're building: ``` What type of data source? ├─ Third-party API (Toast, Square, Stripe, etc.) │ └─ Read: playbook-api-connectors.md │ ├─ Spreadsheet/File (XLSX, CSV, etc.) │ └─ Read: playbook-spreadsheets.md │ └─ Not sure └─ Ask: "What is the data source? An API, a file/spreadsheet, or something else?" ``` ## Warehouse Handling (Defer + Detect) **Do NOT ask about warehouses upfront.** Source code is warehouse-agnostic. 1. **Generate source code immediately** - client.py, source.py, runner.py work with any warehouse 2. **Detect warehouse when needed** - only when generating transforms or DDL: - Check for existing DLT config (`dlt_secrets.toml`, `.dlt/`) - Check `settings.py` or environment variables - Check `pyproject.toml` for destination dependencies 3. **Ask only if undetectable** - when transforms/DDL are needed and no config found Supported warehouses: ClickHouse, Snowflake, MotherDuck/DuckDB, BigQuery ## Output Contract Every ETL pipeline must produce these files: ### For API Connectors ``` etl/ ├── sources// │ ├── __init__.py │ ├── client.py # API client with auth, rate limiting │ └── _source.py # DLT source with resources ├── runners/ │ └── _.py # Pipeline runner with CLI └── transforms/ # Optional post-load transforms └── compute__metrics.py ``` ### For Spreadsheet Ingestion ``` etl/ ├── sources/spreadsheets/ │ ├── __init__.py │ ├── core.py # Shared utilities (if not exists) │ └── templates/ │ ├── __init__.py │ └── .py # Template implementation ├── runners/ │ └── .py # File-based runner with CLI └── transforms/ # Optional post-load transforms ``` ## Decision Tree ``` Is this a new project? ├─ YES → Set up etl/core/ first │ └─ Then: What source type? └─ NO → What source type? ├─ API → Read playbook-api-connectors.md │ └─ Create: client.py, source.py, runner.py └─ Spreadsheet → Read playbook-spreadsheets.md └─ Create: template.py, runner.py ``` ## Core Setup (New Projects Only) If `etl/core/` doesn't exist, create the framework first: ``` etl/ ├── __init__.py ├── core/ │ ├── __init__.py │ ├── pipeline.py # BasePipelineRunner, PipelineConfig │ ├── chunking.py # Date range utilities │ └── cli.py # Logging and CLI helpers ├── sources/ │ └── __init__.py ├── runners/ │ └── __init__.py └── transforms/ └── __init__.py ``` See `templates/core/` for the implementation files. ## Quality Checklist Before marking complete, verify: - [ ] **Structure**: Files in correct directories following conventions - [ ] **Imports**: All imports resolve, no circular dependencies - [ ] **DLT Resources**: Use `@dlt.resource` with `write_disposition` and `primary_key` - [ ] **Incremental Loading**: Cursor fields defined for date-based resources - [ ] **Backfill Mode**: Resources handle backfill (reset cursor, chunking) - [ ] **Error Handling**: Graceful degradation (return empty, don't crash) - [ ] **CLI**: Runner has Typer CLI with run, test, config commands - [ ] **Configuration**: Secrets via env vars or settings module - [ ] **Testing**: Mock client/data available for dry-run mode - [ ] **Makefile**: Add targets for preview/test/run/schema (see etl-style-guide.md) ## Key Patterns ### DLT Resource Pattern ```python @dlt.resource(name="orders", write_disposition="merge", primary_key="id") def orders_resource( modified_date: dlt.sources.incremental[str] = dlt.sources.incremental( "modified_date", initial_value=pendulum.now().subtract(days=7).isoformat() ) ): if backfill_mode: modified_date.start_value = "2015-01-01T00:00:00Z" for entity_id in entity_ids: yield lambda eid=entity_id: _fetch_orders(eid, client) ``` ### Runner Pattern ```python class MyRunner(BasePipelineRunner): @property def pipeline_name(self) -> str: return "my_pipeline" def get_source(self, config, ...): return my_source(...) def get_resources_config(self) -> dict[str, bool]: return {"static_data": False, "time_series": True} ``` ## Reference Files - `etl-style-guide.md` - Naming conventions, directory structure - `warehouse-modeling.md` - DDL patterns for each warehouse - `playbook-api-connectors.md` - Complete API integration guide - `playbook-spreadsheets.md` - Spreadsheet ingestion guide - `templates/` - Copy-paste-ready code templates ## Essential Commands ```bash # Run pipeline uv run python -m etl.runners. run # Test with mock data uv run python -m etl.runners. test # Dry run (DuckDB, no warehouse) uv run python -m etl.runners. run --dry-run # Backfill historical data uv run python -m etl.runners. run --backfill --start-date=2024-01-01 # Show configuration uv run python -m etl.runners. config ``` --- > Converted and distributed by [TomeVault](https://tomevault.io/claim/oxy-hq) — claim your Tome and manage your conversions.