---
name: migrate-ewah-to-dlt-connector
description: Migrate a connector from EWAH (ELT With Airflow Helper) to a standalone dlt connector. Use when replacing a legacy EWAH operator/hook with dlt, analyzing EWAH source code for migration, or converting an EWAHBaseOperator DAG to the dlt-connectors monorepo pattern.
argument-hint: "[connector-name]"
disable-model-invocation: true
---

# Migrate EWAH Connector to dlt

Migrate an API connector from the EWAH framework to a standalone, containerized dlt connector. EWAH is an Apache Airflow-based ELT framework where extraction logic is split between Operators (orchestration) and Hooks (API communication). The goal is to extract the reusable API logic and repackage it as a dlt pipeline.

## Context

EWAH connectors consist of two files:

- **Operator** (`ewah/operators/<source>.py`) - Extraction orchestration, parameters, strategy handling
- **Hook** (`ewah/hooks/<source>.py`) - API communication, authentication, pagination, data fetching

The migration extracts the reusable API logic from these files and creates a standalone dlt connector following the dlt-connectors monorepo pattern. Database connectors (Postgres, MySQL, etc.) should be skipped -- use dlt's built-in `sql_database` source instead.

See `${CLAUDE_SKILL_DIR}/references/ewah-architecture.md` for a comprehensive EWAH technical reference, `${CLAUDE_SKILL_DIR}/references/ewah-to-dlt-mapping.md` for concept mapping, and `${CLAUDE_SKILL_DIR}/references/ewah-connectors-list.md` for the full list of connectors to migrate.

## Prerequisites

Before starting, confirm with the user:

- **Connector name** - `$ARGUMENTS` (which EWAH connector to migrate)
- **Credentials available** - User must provide API credentials for testing
- **Priority resources** - Start with most important endpoints, or migrate all?

## Steps

### Phase 1: Analysis

1. Read the EWAH Operator file (`ewah/operators/<source>.py`) and extract:
   - `_NAMES` (connector aliases used in YAML config)
   - `_ACCEPTED_EXTRACT_STRATEGIES` per resource
   - All `__init__()` parameters with defaults
   - Date filtering logic in `ewah_execute()`

2. Read the EWAH Hook file (`ewah/hooks/<source>.py`) and document:
   - Base URL and endpoint patterns
   - Authentication method (API key, OAuth, token)
   - Pagination pattern (cursor, offset, Link header)
   - ALL resources/objects with their configurations
   - `get_cleaner_callables()` transformations
   - Rate limiting handling

3. Create an implementation plan documenting:
   - All resources to implement
   - Authentication requirements
   - Pagination per resource
   - Write disposition (merge vs replace) based on EWAH extract strategy
   - Incremental cursor fields
   - Data transformations needed

   Ask the user to review and approve before proceeding.

### Phase 2: Implementation

4. Create a feature branch in the dlt-connectors repository.

   ```bash
   cd dlt-connectors
   git checkout main && git pull
   git checkout -b feat/<connector_name>
   ```

5. Create the connector directory structure and required files following the `create-dlt-connector` skill.

   ```bash
   mkdir -p connectors/<connector_name>/.dlt
   ```

   Required files:
   - `<connector_name>_pipeline.py` - Main entry point
   - `pyproject.toml` - Dependencies
   - `.dlt/secrets_example.toml` - Credentials template
   - `.env.example` - Environment variables
   - `.gitignore` - Ignore patterns

6. Implement the source, choosing a pattern based on complexity:

   | Complexity | Pattern |
   |------------|---------|
   | Simple REST API | `rest_api_source` declarative config |
   | Auth + custom config | `@dlt.source` + `rest_api_resources` |
   | Complex logic | Custom source module with `@dlt.source` and `@dlt.resource` |

   See `${CLAUDE_SKILL_DIR}/references/dlt-patterns.md` for implementation templates and `${CLAUDE_SKILL_DIR}/references/ewah-to-dlt-mapping.md` for mapping EWAH concepts to dlt.

7. Test incrementally -- run the pipeline after implementing each resource, not just at the end.

   ```bash
   cd connectors/<connector_name>
   uv sync
   uv run --env-file=.env python <connector_name>_pipeline.py
   ```

8. Log decisions after each significant choice (architecture, scope, trade-offs, test results).

### Phase 3: Finalization

9. Complete the package: lock dependencies and create README.md.

   ```bash
   uv lock
   ```

10. Test with environment variables (same way Airflow will invoke it).

    ```bash
    DLT_DESTINATION=postgres \
    DLT_SOURCE_NAME=<connector_name> \
    uv run python <connector_name>_pipeline.py
    ```

11. Tag the connector version and create a pull request.

    ```bash
    git add .
    git commit -m "feat(<connector_name>): add connector"
    git push -u origin feat/<connector_name>
    gh pr create --base main --title "feat(<connector_name>): add connector"
    ```

    After merge, tag the release:

    ```bash
    git checkout main && git pull
    git tag <connector_name>/v0.1.0
    git push origin <connector_name>/v0.1.0
    ```

12. Update the client Airflow repo to use the new connector (see the `sync-dlt-connectors` and `create-dlt-uv-connector-dag` skills).

## Validation

- [ ] Implementation plan approved by user
- [ ] All EWAH resources migrated (full feature parity)
- [ ] `uv sync` succeeds
- [ ] Pipeline loads data locally
- [ ] Docker build succeeds
- [ ] Docker run loads data
- [ ] README.md complete
- [ ] DECISIONS.md updated
- [ ] PR created and version tagged

## Examples

**Migration rules:**

- One connector at a time with full focus
- Full feature parity -- ALL endpoints/resources from EWAH must be migrated
- Preserve extraction strategy: if EWAH uses incremental, use incremental in dlt
- Skip database connectors -- use dlt's built-in `sql_database` source
- Always use `uv`, never `python` or `pip` directly

**Required dlt settings** (always include in `secrets_example.toml`):

```toml
[schema]
naming = "sql_ci_v1"
json_normalizer = '{"module": "dlt.common.normalizers.json.relational", "config": {"max_nesting": 0}}'
```

- `naming = "sql_ci_v1"` ensures case-insensitive, consistent table/column names
- `max_nesting: 0` prevents child tables for nested JSON (one table per resource)