---
name: create-airflow3-connector-dag
description: Create an Airflow 3 @task DAG that runs a dlt connector via uv subprocess with credentials from Airflow Connections. Use when writing a DAG for a connector in the connectors/ folder of an Airflow 3 template repo, wiring per-resource task fan-out, or passing source credentials from a Connection to a dlt subprocess.
disable-model-invocation: true
argument-hint: "[connector-name]"
---

# Create an Airflow 3 Connector DAG

Create a DAG for the `$ARGUMENTS` connector in an Airflow 3 template repository, using the `@task` decorator and the shared `dags/utils/dlt.py` utilities.

## Context

Connectors live as hard copies under `connectors/<connector_name>/`, bind-mounted to `/opt/airflow/connectors/`. `run_dlt_connector()` runs `uv run --no-dev python <connector_name>_pipeline.py` in an isolated per-connector venv, streams output to the task log, kills the process tree on interruption, and deletes the run's `DLT_PIPELINES_DIR` after success (dlt state lives in the destination and is restored next run; on failure the dir is kept so retries resume pending load packages).

For Airflow 2.x / EWAH, use `create-dlt-uv-connector-dag` instead. Exception: repos set up with the **modern hard-copy variant** of `setup-dlt-uv-connector-airflow` (Airflow 2 + the template's `utils/dlt.py`) use the patterns in THIS skill — the utilities are Airflow-2-compatible, only the DAG imports differ: swap `from airflow.sdk import Variable, dag, task` for `from airflow.decorators import dag, task` / `from airflow.models import Variable`. There is no `@task.subprocess` in any Airflow version; the subprocess runs inside a regular task via `run_dlt_connector()`.

## Prerequisites

- Airflow 3 template set up (see `setup-airflow3-dlt-template`)
- The connector present in `connectors/<connector_name>/` — if missing, add it with `update-airflow3-connectors` first
- The connector's pipeline honors the standard env vars (`DLT_DESTINATION`, `DLT_SOURCE_NAME`, `DLT_DATASET_NAME`, `DLT_PIPELINES_DIR`)

**Understand the connector before writing the DAG** — read its local copy in this order (later sources are authoritative over earlier ones being stale):

1. `connectors/<connector_name>/README.md` — intended usage, credentials, resources
2. `connectors/<connector_name>/<connector_name>_pipeline.py` (+ source module if present) — the **actual** env vars: `grep os.getenv` here; READMEs can be thin or stale, the source cannot
3. `connectors/<connector_name>/.dlt/*_example.toml` — the required `sources.*` secret/config keys, which map 1:1 to `SOURCES__*` env vars

If the local docs are thin or look stale, consult the **dlt-connectors monorepo**: the connector's upstream README and the repo's `CLAUDE.md` (connector conventions: env-var pattern, incremental loading, schema contracts, `DLT_SOURCE_NAME` ↔ secrets-section mapping). Remember the local copy may carry deliberate customizations — for behavior, local source wins over upstream docs.

**If the credential shape or resource selection is still unclear after reading, ask the user before proceeding.**

**Don't re-derive repo conventions** — the DAG template and credential patterns embedded below ARE the conventions; write from them directly instead of re-reading the example DAGs and utils first. Specifically: do NOT open `utils/dlt.py`, `utils/notifications.py`, or any existing DAG "to check signatures" or "match conventions" — their interfaces (`get_destination_env(destination)`, `get_sql_source_env(conn_id, source_name)`, `run_dlt_connector(name, env)`, `notify_slack_on_failure`) are stable and exactly as used in the template below. Read the repo only AFTER something contradicts (an import error at validation time). One subtlety that DOES require the connector source (covered by the reading order above): for decorator-injected credentials (`dlt.secrets.value` args), dlt derives the config section from the **module that defines the `@dlt.source` function** (the package name when it lives in `<pkg>/__init__.py`; overridable via `section=` on the decorator) — NOT from `DLT_SOURCE_NAME` and NOT from the function name (that's only a deeper optional level, `SOURCES__<module>__<function>__*`). Connector READMEs get this wrong (personio documented a section dlt never searches). Two exceptions/tools: connectors doing explicit `dlt.secrets["sources.…"]` lookups (sql_database) use exactly the path their code names; and on any credentials miss, dlt's `ConfigFieldMissingException` lists the exact env keys it searched — treat that list as ground truth over any documentation.

## Steps

### 1. Pick the DAG shape from the three bundled examples

| Example DAG | Shape | Use when |
|---|---|---|
| `dag_extract_load__pokemon.py` | Single task, no source credentials | Public APIs, resources hardcoded in the connector |
| `dag_extract_load__fx.py` | One task per resource (env-var fan-out) | Parallel per-resource loads with parameterized endpoints |
| `dag_extract_load__sql_database.py` | Source credentials from an Airflow Connection | Any source that needs credentials |

### 2. Create `dags/dag_extract_load__<connector_name>.py`

Minimal pattern (all shapes build on this):

```python
import pendulum
from airflow.sdk import Variable, dag, task
from airflow.operators.empty import EmptyOperator

from utils.dlt import get_destination_env, run_dlt_connector
from utils.notifications import notify_slack_on_failure

DEFAULT_ARGS = {
    "retries": 2,
    "retry_delay": pendulum.duration(minutes=1),
    "on_failure_callback": notify_slack_on_failure,
}


@dag(
    schedule=None,  # set a cron once verified
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args=DEFAULT_ARGS,
    tags=["extract_load", "dlt", "<connector_name>"],
)
def extract_load_<connector_name>():
    """Document here: credential setup, endpoint selection, Connection IDs."""

    start = EmptyOperator(task_id="start")
    end = EmptyOperator(task_id="end")

    @task()
    def load():
        destination = Variable.get("dlt_destination", default="postgres")
        env = {
            **get_destination_env(destination),
            "DLT_SOURCE_NAME": "<connector_name>",
            "DLT_DATASET_NAME": "<connector_name>",
            "DLT_PIPELINES_DIR": "/tmp/dlt_pipelines/<connector_name>",
        }
        run_dlt_connector("<connector_name>", env)

    start >> load() >> end


dag = extract_load_<connector_name>()
```

**Always pass a unique `DLT_PIPELINES_DIR`** — without it, dlt writes state into the mounted `~/.dlt/pipelines` (the repo's `.dlt/` on the host), and the automatic post-success cleanup cannot run.

### 3. Wire source credentials (if the source needs them)

**Database sources** — use `get_sql_source_env()`; it reads an Airflow Connection and emits the whole credential set as one `SOURCES__<SOURCE_NAME>__CREDENTIALS` env var (JSON object — dlt's env provider parses it into the credentials dict):

```python
from utils.dlt import get_sql_source_env

env = {
    **get_destination_env(destination),
    **get_sql_source_env("<conn_id>", "<source_name>"),
    ...
}
```

**API-token sources** — read the token from a Connection field and set the connector's documented env var:

```python
from airflow.hooks.base import BaseHook

try:
    conn = BaseHook.get_connection("<connector_name>")
    if conn.password:  # empty string would be a *present* env var and mask the fallback
        # <source_module> = the module defining the @dlt.source function — see
        # the prefix rule in Prerequisites; NOT DLT_SOURCE_NAME, NOT the function name
        env["SOURCES__<SOURCE_MODULE>__API_TOKEN"] = conn.password
except Exception:
    pass  # no Connection — dlt falls back to secrets.toml
```

Both helpers return/degrade to nothing when the Connection is absent → dlt falls back to the mounted `secrets.toml`. Document the Connection ID and field mapping in the DAG docstring — the docstring is the operator's setup guide.

**Scaffold the Connection (ask the user first).** Ask which credential path the user wants:

- **UI Connection** → create the Connection scaffold now with unmistakable placeholders, so the user only edits values in Admin → Connections instead of building the field mapping from docs:

  ```bash
  # never overwrite: check first
  docker compose exec airflow-scheduler airflow connections get <conn_id> >/dev/null 2>&1 \
    && echo "exists - leaving untouched" \
    || docker compose exec airflow-scheduler airflow connections add <conn_id> \
         --conn-type <postgres|generic> \
         --conn-host REPLACE_ME --conn-port 5432 --conn-schema REPLACE_ME \
         --conn-login REPLACE_ME --conn-password REPLACE_ME
  ```

  (API-token sources usually need only `--conn-type generic --conn-password REPLACE_ME`.) Tell the user explicitly: **edit the placeholders in the UI before the first run** — placeholder values fail at connect time, by design.

- **secrets.toml** → do **NOT** create the Connection. Its mere existence overrides the TOML fallback (placeholder values would mask working file-based credentials), and for DB sources an empty scaffold trips the incomplete-Connection validation on every run.

### 4. Per-resource fan-out (parallel loads)

One task per resource, each with its own `DLT_SOURCE_NAME` and `DLT_PIPELINES_DIR` so parallel runs never share pipeline state:

```python
RESOURCES = ["<resource_a>", "<resource_b>"]

@task()
def load_resource(resource_name: str):
    destination = Variable.get("dlt_destination", default="postgres")
    env = {
        **get_destination_env(destination),
        "<CONNECTOR_RESOURCES_ENV>": resource_name,
        "DLT_SOURCE_NAME": f"<connector_name>_{resource_name}",
        "DLT_PIPELINES_DIR": f"/tmp/dlt_pipelines/<connector_name>_{resource_name}",
        "DLT_DATASET_NAME": "<connector_name>",
    }
    run_dlt_connector("<connector_name>", env)

for resource_name in RESOURCES:
    t = load_resource.override(task_id=f"load_{resource_name}")(resource_name)
    start >> t >> end
```

**Fan-out requires the connector to honor `DLT_PIPELINES_DIR`** (the prerequisite env-var pattern). A hard-copied connector lacking it shares `~/.dlt/pipelines/<pipeline_name>` across all runs — parallel tasks with the same pipeline name then corrupt dlt local state. Fix the connector (add the pattern, contribute it upstream); only if it must stay unmodified, chain such tasks sequentially instead of fanning out.

### 5. Optionally register in the orchestrator

Append the DAG ID to `EXTRACT_LOAD_DAG_IDS` in `dags/dag_orchestrate.py` to include it in the scheduled EL → dbt chain.

### 6. Commit and open a PR

**If the connector was added earlier in this session** (via `update-airflow3-connectors` with the commit deferred), commit connector + DAG **together** — they are one logical change; don't leave the earlier add orphaned or split it into two PRs.

Never commit to `main` — work on a feature branch. After the Validation checklist passes and **after confirming with the user**:

```bash
git checkout -b feat/dag-<connector_name>   # skip if already on one
git add dags/
git commit -m "feat: add extract_load_<connector_name> DAG"
git push -u origin feat/dag-<connector_name>
gh pr create --fill
```

For broader end-of-session git wrap-up, see the `wrap` skill in the `gemma-devx` plugin.

## Validation

- [ ] `docker compose exec airflow-scheduler airflow dags list-import-errors` — no errors
- [ ] Trigger the DAG; task logs show dlt load info / row counts
- [ ] With the Connection deleted, the DAG still works via `secrets.toml` fallback (if configured)
- [ ] `/tmp/dlt_pipelines/` inside the scheduler container is empty after a successful run

## Troubleshooting

| Issue | Solution |
|---|---|
| Retries keep failing on local package/state errors | Rare corrupted working dir: `docker compose exec airflow-scheduler rm -rf /tmp/dlt_pipelines/<source_name>` and re-trigger |
| First run very slow | uv installing the connector venv (one-time per container lifetime); cached at `/tmp/dlt_venvs/<connector_name>` afterwards — do not clean per-run |
| State appears in the repo's `.dlt/pipelines/` on the host | The DAG task is missing `DLT_PIPELINES_DIR` |
| Credentials not picked up | Precedence is UI Connection > container env > secrets.toml; check Connection ID spelling. On a `ConfigFieldMissingException`, read the **searched-keys list in the error** — it names the exact section dlt expects (module-derived; connector READMEs can be stale) |
| `uv: command not found` | Image not built from the template Dockerfile — `docker compose build` |

**Related skills:** `update-airflow3-connectors` (copy/update connectors) · `airflow3-dlt-credentials` (credential model) · `create-dlt-uv-connector-dag` (Airflow 2/EWAH generation)