# DLT Connector Framework Specification

This document provides a comprehensive specification for creating new dlt connectors within a dlt-connectors monorepo. It describes the code organization, patterns, and containerization requirements.

---

## Repository Structure Overview

```
dlt-connectors/
├── base_images/                 # Docker base images (shared)
│   ├── python_base/             # Python 3.12 + uv
│   └── dlt_base/                # python_base + entrypoint for secrets handling
├── connectors/                  # Individual connector directories
│   └── <connector_name>/        # Each connector is self-contained
├── utils/                       # Shared utilities
├── Justfile                     # Build automation commands
└── pyproject.toml               # Root-level dev dependencies
```

---

## Connector Directory Structure

### Minimal Structure (Simple Connector)

```
<connector_name>/
├── Dockerfile                   # REQUIRED: Container definition
├── <connector_name>_pipeline.py # REQUIRED: Main entry point
├── pyproject.toml               # REQUIRED: Dependencies
├── uv.lock                      # REQUIRED: Lock file for reproducibility
├── .env.example                 # REQUIRED: Environment variable template
├── .dlt/
│   └── secrets_example.toml     # REQUIRED: Credentials template (copy to secrets.toml)
├── .gitignore                   # REQUIRED: Ignore .dlt/secrets.toml, .env, etc.
└── README.md                    # REQUIRED: Usage documentation
```

### Extended Structure (Complex Connector)

```
<connector_name>/
├── Dockerfile                   # REQUIRED
├── <connector_name>_pipeline.py # REQUIRED
├── pyproject.toml               # REQUIRED
├── uv.lock                      # REQUIRED
├── .env.example                 # REQUIRED
├── .dlt/
│   └── secrets_example.toml     # REQUIRED
├── .gitignore                   # REQUIRED
├── README.md                    # REQUIRED
├── <connector_name>/            # OPTIONAL: Custom source module
│   ├── __init__.py              # Source definition with @dlt.source
│   └── helpers/                 # Helper utilities
├── settings.py                  # OPTIONAL: Runtime configuration
├── transformations.py           # OPTIONAL: Data transformation functions
├── config.py                    # OPTIONAL: Configuration classes
└── tests/                       # OPTIONAL: Connector-specific tests
```

---

## Pipeline Entry Point Pattern

Every connector MUST follow this pattern in its main pipeline file:

### Required Environment Variables

```python
import os

# REQUIRED: These two environment variables must be defined at module level
DLT_DESTINATION = os.getenv("DLT_DESTINATION", "duckdb")  # Default fallback
DLT_SOURCE_NAME = os.getenv("DLT_SOURCE_NAME", "<connector_name>")
```

### Pipeline Configuration

```python
import dlt

def run_pipeline():
    pipeline = dlt.pipeline(
        pipeline_name=DLT_SOURCE_NAME,      # Use env var
        destination=DLT_DESTINATION,         # Use env var
        dataset_name=DLT_SOURCE_NAME,        # Typically matches source name
        dev_mode=False,                      # Set to False for production
    )

    source = your_source_function()
    load_info = pipeline.run(source)
    print(load_info)
    return load_info

if __name__ == "__main__":
    run_pipeline()
```

---

## Source Implementation Patterns

The framework supports three source patterns, from simplest to most complex:

### Pattern 1: Declarative REST API Source (Simplest)

Use `rest_api_source` for straightforward REST APIs:

```python
from dlt.sources.rest_api import rest_api_source

pokemon_source = rest_api_source({
    "client": {
        "base_url": "https://api.example.com/",
    },
    "resource_defaults": {
        "endpoint": {"params": {"limit": 1000}},
        "write_disposition": "replace",
    },
    "resources": [
        "resource1",
        "resource2",
    ],
})
```

### Pattern 2: Custom REST API Source with @dlt.source

For APIs requiring authentication or complex configurations:

```python
import dlt
from dlt.sources.rest_api import RESTAPIConfig, rest_api_resources

@dlt.source
def my_source(api_token: str = dlt.secrets.value):
    config: RESTAPIConfig = {
        "client": {
            "base_url": "https://api.example.com/",
            "auth": {
                "type": "api_key",
                "name": "api_token",
                "api_key": api_token,
                "location": "query",  # or "header"
            },
        },
        "resource_defaults": {
            "primary_key": "id",
            "write_disposition": "merge",
        },
        "resources": [
            {
                "name": "resource_name",
                "endpoint": {
                    "path": "v1/resource",
                    "method": "GET",
                    "data_selector": "data",
                    "paginator": {...},
                    "incremental": {...},
                },
            },
        ],
    }
    yield from rest_api_resources(config)
```

### Pattern 3: Fully Custom Source Module (Most Flexible)

For complex APIs requiring custom logic, create a source module:

**Directory structure:**

```
<connector_name>/
├── <connector_name>/
│   ├── __init__.py          # Contains @dlt.source and @dlt.resource
│   └── helpers/
│       └── data_processing.py
├── <connector_name>_pipeline.py
└── settings.py
```

**Source module (`<connector_name>/__init__.py`):**

```python
import dlt
from dlt.sources import DltResource
from typing import Iterator, List

@dlt.source()
def my_source(source_name: str) -> List[DltResource]:
    credentials = dlt.secrets[f"sources.{source_name}.credentials"]

    return [
        my_resource(credentials, source_name),
    ]

@dlt.resource(
    primary_key=["id", "date"],
    write_disposition="merge",
)
def my_resource(
    credentials,
    source_name: str,
    cursor: str = dlt.sources.incremental("date", initial_value="2024-01-01"),
) -> Iterator[dict]:
    for record in fetch_data(credentials, cursor.last_value):
        yield record
```

---

## Environment Variables and Secrets

### Understanding the Separation

| Type | Storage | Contains | Git Committed |
|------|---------|----------|---------------|
| **Environment Variables** | `.env` file | Runtime config (destination, source name) | No (use `.env.example`) |
| **Secrets** | `.dlt/secrets.toml` | Credentials (API keys, passwords, tokens) | No (use `.dlt/secrets_example.toml`) |
| **Config** | `.dlt/config.toml` | Non-sensitive settings | Optional |

### Standard Environment Variables

| Variable | Required | Description |
|----------|----------|-------------|
| `DLT_DESTINATION` | Yes | Target destination (postgres, bigquery, snowflake, duckdb) |
| `DLT_SOURCE_NAME` | Yes | Identifies the source in secrets.toml and names the dataset |

### Accessing Secrets in Code

dlt automatically injects secrets into `@dlt.source` function parameters:

```python
@dlt.source
def my_source(
    api_key: str = dlt.secrets.value,      # Required - from [sources.<name>]
    api_secret: str = dlt.secrets.value,   # Required - from [sources.<name>]
    optional_setting: str = "default",      # Optional with default
):
    # dlt reads from secrets.toml based on DLT_SOURCE_NAME
    # If DLT_SOURCE_NAME=my_source, reads from [sources.my_source]
    ...
```

### Secrets Handling in Airflow

When running via Airflow, the `run_dlt_connector()` utility (from `dags/utils/common.py`) handles secrets:

1. Reads raw TOML content from the `dlt_secrets_toml` Airflow Variable
2. Writes it to the connector's `.dlt/secrets.toml`
3. Runs `uv run python <connector_name>_pipeline.py` in a subprocess

---

## Dynamic Source Name Pattern

For connectors supporting multiple instances (e.g., multiple accounts), use dynamic credential lookup:

```python
import dlt

DLT_SOURCE_NAME = os.getenv("DLT_SOURCE_NAME", "default_source")

# In source/resource functions:
credentials = dlt.secrets[f"sources.{source_name}.credentials"]
```

This allows the same connector image to work with different credentials by changing `DLT_SOURCE_NAME`.

---

## Multi-File Source Module Pattern

When a connector needs helper functions or complex logic:

### Directory Structure

```
connector/
├── connector/
│   ├── __init__.py           # Main source with @dlt.source
│   ├── helpers/
│   │   ├── __init__.py
│   │   └── data_processing.py
│   └── setup_script.py       # Optional setup utilities
├── settings.py               # Runtime configuration from env vars
└── connector_pipeline.py     # Entry point
```

### settings.py Pattern

```python
import os
from datetime import datetime

DEFAULT_START_DATE = datetime.strptime(
    os.getenv("DLT_CONNECTOR_DEFAULT_START_DATE", "2024-01-01"),
    "%Y-%m-%d"
).date()
```

---

## README.md Template

Every connector MUST have a `README.md` with these sections:

```markdown
# <Connector Name>

Brief description of what this connector does and what API/data source it connects to.

## Resources

| Name | Description |
|------|-------------|
| resource_1 | Description of what this resource contains |

## Credentials

This connector requires the following credentials in `.dlt/secrets.toml`:

| Field | Description |
|-------|-------------|
| `api_key` | Your API key |

## Configuration

1. Copy the secrets example file:
   ```bash
   cp .dlt/secrets_example.toml .dlt/secrets.toml
   ```
2. Copy the environment example file:
   ```bash
   cp .env.example .env
   ```
3. Update `.dlt/secrets.toml` with your credentials
4. Update `.env` with your configuration

## Running with Python

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

## Running Locally

```bash
# From connector directory
cd connectors/<connector_name>
uv sync
DLT_DESTINATION=postgres DLT_SOURCE_NAME=<source_name> uv run python <connector_name>_pipeline.py

# Or from repo root via just
just run <connector_name>
```

---

## Checklist for Creating a New Connector

### Required Files

- [ ] `<connector_name>_pipeline.py` - Entry point with `DLT_DESTINATION`, `DLT_SOURCE_NAME`
- [ ] `pyproject.toml` + `uv.lock` - Dependencies
- [ ] `.env.example` - Template with required environment variables
- [ ] `.dlt/secrets_example.toml` - Template with required dlt settings and credentials structure
- [ ] `.gitignore` - Ignores `.dlt/secrets.toml`, `.env`, `__pycache__/`
- [ ] `README.md` - Documentation with overview, resources, configuration, and run commands

### Pipeline Code Requirements

- [ ] Define `DLT_DESTINATION = os.getenv("DLT_DESTINATION", "duckdb")` at module level
- [ ] Define `DLT_SOURCE_NAME = os.getenv("DLT_SOURCE_NAME", "<name>")` at module level
- [ ] Use these variables in `dlt.pipeline()` call
- [ ] Include `if __name__ == "__main__":` block
- [ ] Print `load_info` for logging