---
name: create-dlt-uv-connector-dag
description: Create an Airflow DAG that runs dlt connectors via PythonOperator with uv subprocess isolation. Use when creating a DAG for a synced connector, setting up run_dlt_connector() tasks, configuring multi-instance extraction, or building the uv-based alternative to DockerOperator DAGs.
disable-model-invocation: true
---

# Create a uv Connector DAG

Create an Airflow DAG that runs dlt connectors via `uv` using `PythonOperator` and the shared `run_dlt_connector` utility. Each connector runs in its own isolated virtual environment, managed by `uv`, without requiring Docker-in-Docker infrastructure.

## Context

Connectors are synced from the dlt-connectors monorepo into `connectors/` and mounted into Airflow at `/opt/airflow/connectors/`. The `run_dlt_connector()` utility from `dags/utils/common.py`:

1. Reads secrets from the `dlt_secrets_toml` Airflow Variable
2. Writes `secrets.toml` directly into the connector's `.dlt/` directory
3. Runs `uv run python <connector_name>_pipeline.py` in a subprocess

This pattern is compatible with all Airflow 2.x versions (including EWAH with Python 3.8).

## Prerequisites

- Airflow set up for uv connectors (see `setup-dlt-uv-connector-airflow` skill)
- Connector synced to `connectors/<connector_name>/`
- Shared utilities at `dags/utils/common.py` (see `setup-dlt-uv-connector-airflow` skill, step 9)
- Airflow Variables configured:
  - `dlt_destination` — target destination (e.g., `postgres`, `snowflake`)
  - `dlt_secrets_toml` — full TOML content of `secrets.toml`

## Steps

1. Ensure the shared utilities module exists at `dags/utils/common.py` with `run_dlt_connector()`. The full template is in the `setup-dlt-uv-connector-airflow` skill (step 9).

2. Create a new DAG file at `dags/load/dag_<connector_name>.py`.

3. Define the DAG with imports.

   ```python
   import pendulum

   from airflow.decorators import dag
   from airflow.operators.python import PythonOperator

   from utils.common import run_dlt_connector
   ```

4. Create the DAG using the `@dag` decorator.

   ```python
   @dag(
       schedule_interval=None,  # Set schedule as needed (e.g., "@daily")
       start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
       catchup=False,
       tags=["extract_load", "dlt"],
   )
   def extract_load_<connector_name>():
       ...
   ```

5. Create a task using `PythonOperator` that runs the connector.

   ```python
   PythonOperator(
       task_id="load_<connector_name>",
       python_callable=run_dlt_connector,
       op_kwargs={
           "connector_name": "<connector_name>",
           "source_name": "<source_name>",
       },
   )
   ```

6. Instantiate the DAG.

   ```python
   dag = extract_load_<connector_name>()
   ```

7. For connectors with multiple instances (e.g., multiple ad accounts), create separate tasks with different `source_name` values. Each `source_name` maps to a separate section in `secrets.toml` for per-instance credentials.

   ```python
   @dag(
       schedule_interval=None,
       start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
       catchup=False,
       tags=["extract_load", "dlt"],
   )
   def extract_load_google_ads():

       PythonOperator(
           task_id="load_google_ads_germany",
           python_callable=run_dlt_connector,
           op_kwargs={
               "connector_name": "google_ads",
               "source_name": "google_ads_germany",
           },
       )

       PythonOperator(
           task_id="load_google_ads_austria",
           python_callable=run_dlt_connector,
           op_kwargs={
               "connector_name": "google_ads",
               "source_name": "google_ads_austria",
           },
       )

   dag = extract_load_google_ads()
   ```

8. For task dependencies, assign operators to variables and use `>>`.

   ```python
   load_germany = PythonOperator(
       task_id="load_google_ads_germany",
       python_callable=run_dlt_connector,
       op_kwargs={"connector_name": "google_ads", "source_name": "google_ads_germany"},
   )

   load_austria = PythonOperator(
       task_id="load_google_ads_austria",
       python_callable=run_dlt_connector,
       op_kwargs={"connector_name": "google_ads", "source_name": "google_ads_austria"},
   )

   load_germany >> load_austria   # sequential
   ```

9. For passing additional environment variables (e.g., resource configs for `sql_database`), use the `extra_env` parameter.

   ```python
   import json

   RESOURCES_CONFIG = [
       {"table": "users", "schema": "public", "incremental": "modified"},
   ]

   PythonOperator(
       task_id="load_sql_database",
       python_callable=run_dlt_connector,
       op_kwargs={
           "connector_name": "sql_database",
           "source_name": "my_database",
           "extra_env": {
               "DLT_RESOURCES_CONFIG": json.dumps(RESOURCES_CONFIG),
           },
       },
   )
   ```

## Validation

- [ ] DAG appears in Airflow UI without import errors
- [ ] Task runs successfully (subprocess returns exit code 0)
- [ ] Data loads to the expected destination
- [ ] Logs show connector stdout/stderr in Airflow task logs

## Examples

**Complete working DAG (tested with EWAH + Python 3.8):**

```python
"""
Google Ads extract & load DAG — runs dlt connector via uv.

Each task runs the google_ads connector with a different DLT_SOURCE_NAME,
which maps to a separate section in secrets.toml for per-account credentials.
"""

import pendulum

from airflow.decorators import dag
from airflow.operators.python import PythonOperator

from utils.common import run_dlt_connector


@dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2025, 7, 27, tz="UTC"),
    catchup=False,
    tags=["extract_load", "dlt"],
)
def extract_load_google_ads():

    PythonOperator(
        task_id="load_google_ads_germany",
        python_callable=run_dlt_connector,
        op_kwargs={
            "connector_name": "google_ads",
            "source_name": "google_ads_germany",
        },
    )

    PythonOperator(
        task_id="load_google_ads_austria",
        python_callable=run_dlt_connector,
        op_kwargs={
            "connector_name": "google_ads",
            "source_name": "google_ads_austria",
        },
    )


dag = extract_load_google_ads()
```

**DAG naming convention:** `dags/load/dag_<connector_name>.py` for EL pipelines.

**Comparison with DockerOperator approach:**

| Aspect | uv / PythonOperator | DockerOperator |
|--------|---------------------|----------------|
| Isolation | Virtual environment (uv) | Container |
| Infrastructure | None extra | docker-proxy + socket mount |
| Startup time | ~1-2s | ~5-10s |
| Debugging | Native Airflow logs | Container logs |
| Airflow compatibility | All 2.x versions | Requires docker provider package |

**Troubleshooting:**

| Issue | Solution |
|-------|----------|
| `uv: command not found` | Add `RUN pip install uv` to Airflow Dockerfile. Check PATH in `run_dlt_connector()` |
| `Permission denied: .dlt/secrets.toml` | Re-run `./sync-connectors.sh` — it sets `chmod -R a+w` on connector dirs |
| `Python >=3.12 required` | uv auto-downloads Python; ensure internet access from Airflow container |
| `CalledProcessError` / non-zero exit | Check Airflow task logs — stdout/stderr are printed before the error |
| `Variable dlt_secrets_toml not found` | Set the variable: `airflow variables set dlt_secrets_toml "$(cat secrets.toml)"` |
| `ModuleNotFoundError: tomllib` | Don't import tomllib — use the common.py template which avoids it (Python 3.8 safe) |
| `@task.bash not found` | Don't use task decorators — use `PythonOperator` with `run_dlt_connector()` |
| `append_env` invalid argument | Don't use `BashOperator` with `append_env=True` — requires Airflow 2.4+ |