# EWAH Framework Architecture

## Overview

EWAH (ELT With Airflow Helper) is an Apache Airflow-based ELT framework that extracts data from various sources and loads it into relational data warehouses (PostgreSQL, Snowflake, BigQuery).

### Core Philosophy

- **ELT-focused**: Raw data extraction and loading only; no transformations beyond what's needed for relational format
- **Airflow-orchestrated**: DAGs manage scheduling, backfill, and incremental loading
- **Schema isolation**: Each source loads into its own schema (e.g., `raw_salesforce`)
- **Atomic operations**: Data loads into a temporary schema (`_next` suffix) then replaces production data

---

## Architecture

### Component Hierarchy

```
DAG Factory (dags.yml / Python)
    └── Operator (extraction + orchestration logic)
            └── Hook (connection + API communication)
                    └── Uploader (destination handling - OUT OF SCOPE)
```

### Key Directories

| Path | Purpose |
|------|---------|
| `ewah/operators/` | Source-specific extraction logic + parameter handling |
| `ewah/hooks/` | Connection management + API communication |
| `ewah/dag_factories/` | DAG generation patterns (atomic, idempotent, mixed) |
| `ewah/constants/` | Shared constants (DWH engines, strategies) |

---

## How to Analyze a Source Connector

### Step 1: Identify the Files

Each connector has up to two files:

- **Operator**: `ewah/operators/<source>.py` - Contains extraction orchestration
- **Hook**: `ewah/hooks/<source>.py` - Contains API/connection logic

### Step 2: Understand the Operator

Look for these key elements:

```python
class EWAHExampleOperator(EWAHBaseOperator):
    _NAMES = ["example", "ex"]  # YAML aliases

    _ACCEPTED_EXTRACT_STRATEGIES = {
        EC.ES_FULL_REFRESH: True,
        EC.ES_INCREMENTAL: True,
        EC.ES_SUBSEQUENT: True,
    }

    def __init__(self, api_resource=None, custom_filter=None, *args, **kwargs):
        self.api_resource = api_resource or kwargs["target_table_name"]
        super().__init__(*args, **kwargs)

    def ewah_execute(self, context):
        for batch in self.source_hook.get_data(...):
            self.upload_data(batch)
```

### Step 3: Understand the Hook

```python
class EWAHExampleHook(EWAHBaseHook):
    _ATTR_RELABEL = {"api_key": "password"}
    conn_type = "ewah_example"

    def get_data(self, resource, filters, data_from, data_until):
        # Makes API calls, handles pagination, rate limiting
        # Returns/yields data as list of dicts
        pass
```

### Step 4: Trace the Data Flow

```
1. YAML Config
   ↓
2. DAG Factory creates Operator with config as kwargs
   ↓
3. Operator.__init__() stores source-specific params
   ↓
4. Airflow calls Operator.execute(context)
   ↓
5. Base execute() sets up self.source_hook, self.data_from, self.data_until
   ↓
6. Base execute() calls ewah_execute(context)
   ↓
7. ewah_execute() calls self.source_hook.get_data(...)
   ↓
8. Hook makes API calls, handles pagination, returns data
   ↓
9. ewah_execute() calls self.upload_data(batch)
```

### Step 5: Find Source-Specific Logic

| What to Find | Where to Look |
|--------------|---------------|
| Required parameters | Operator `__init__()` signature |
| Default values | Operator `__init__()` body |
| API endpoints | Hook class constants or methods |
| Pagination logic | Hook's `get_data()` or `get_data_in_batches()` |
| Authentication | Hook's `__init__()` or property methods |
| Rate limiting | Hook's data fetching methods |
| Data transformation | Hook's `get_cleaner_callables()` |
| Date filtering | Operator's `ewah_execute()` method |

---

## Extract Strategies

| Strategy | Constant | Behavior |
|----------|----------|----------|
| Full Refresh | `EC.ES_FULL_REFRESH` | Load all data every run |
| Incremental | `EC.ES_INCREMENTAL` | Load data within `data_interval_start` to `data_interval_end` |
| Subsequent | `EC.ES_SUBSEQUENT` | Load data newer than the max value of `subsequent_field` |

### Default Load Strategy per Extract Strategy

```python
DEFAULT_LS_PER_ES = {
    ES_FULL_REFRESH: LS_INSERT_REPLACE,
    ES_SUBSEQUENT: LS_UPSERT,
    ES_INCREMENTAL: LS_UPSERT,
}
```

---

## Connector Walkthrough: Shopify

### Operator (`operators/shopify.py`)

```python
class EWAHShopifyOperator(EWAHBaseOperator):
    _NAMES = ["shopify"]
    _ACCEPTED_EXTRACT_STRATEGIES = {
        EC.ES_FULL_REFRESH: True,
        EC.ES_INCREMENTAL: True,
        EC.ES_SUBSEQUENT: True,
    }
```

**Parameters**: `shopify_object`, `shop_id`, `filter_fields`, `api_version`, `get_transactions_with_orders`, `get_events_with_orders`, `get_inventory_data_with_product_variants`

**Extraction logic** (`ewah_execute`):

1. Handle subsequent loading (get max timestamp from existing data)
2. Add metadata (shop_id to each row)
3. Call hook to fetch data with date filters
4. Upload batches

### Hook (`hooks/shopify.py`)

**Authentication**: Access token in password field, shop subdomain in login field

**Object configuration**:

```python
_OBJECTS = {
    "orders": {},  # Uses defaults
    "payouts": {
        "_timestamp_fields": ("date_min", "date_max", "date"),
        "_datetime_format": "%Y-%m-%d",
        "_object_url": "shopify_payments/payouts",
    },
    "balance_transactions": {
        "_is_drop_and_replace": True,
        "_object_url": "shopify_payments/balance/transactions",
    },
}
```

**Pagination**: Link header with `rel="next"` URL

---

## Connection Handling

### Airflow Connection Fields

| Airflow Field | Common Usage |
|---------------|--------------|
| `host` | API endpoint / hostname |
| `login` | Username / API ID |
| `password` | Password / API key / token |
| `schema` | Database name / Account ID |
| `port` | Port number |
| `extra` | JSON with additional fields |

---

## What to Extract vs. Discard

### Extract (Reusable Logic)

From Hooks:
- API communication (base URLs, endpoints, requests)
- Authentication (credentials, headers, tokens)
- Pagination patterns
- Rate limiting handling
- Resource configuration (`_OBJECTS`, `ACCEPTED_OBJECTS`)
- Data transformations (`get_cleaner_callables()`)

From Operators:
- Date filtering parameters
- Incremental logic
- Resource-specific parameters

### Discard (Airflow-Specific)

| EWAH Pattern | Migration Approach |
|--------------|-------------------|
| `context["data_interval_start"]` | Pass as CLI argument or env var |
| `self.log.info()` | Standard Python logging |
| `self.upload_data(data)` | dlt pipeline handles loading |
| `self.get_max_value_of_column()` | dlt incremental state |
| Airflow connections | dlt secrets (`.dlt/secrets.toml`) |
| `self.source_hook` | Direct instantiation |
| DAG factory logic | Not needed |
| YAML configuration parsing | Not needed |

---

## YAML Configuration Example

```yaml
EL_Shopify:
  dag_strategy: incremental
  el_operator: shopify
  target_schema_name: raw_shopify
  operator_config:
    general_config:
      source_conn_id: shopify_connection
      extract_strategy: subsequent
    tables:
      orders:
        get_transactions_with_orders: true
      products:
        get_inventory_data_with_product_variants: true
      customers: {}
```