# DLT Patterns Reference

## Required Files Per Connector

```
connectors/<connector_name>/
├── .dlt/
│   └── secrets_example.toml      # Credentials template
├── Dockerfile                    # Uses dlt-connectors/dlt-base:latest
├── <connector_name>_pipeline.py  # Main entry point
├── pyproject.toml                # Dependencies with dlt
├── uv.lock                       # Lock file
├── .env.example                  # Environment variable template
├── .gitignore                    # Ignores .dlt/, .env, __pycache__/
└── README.md                     # Usage documentation
```

## Pipeline Entry Point Pattern

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

DLT_DESTINATION = os.getenv("DLT_DESTINATION", "duckdb")
DLT_SOURCE_NAME = os.getenv("DLT_SOURCE_NAME", "<connector_name>")

@dlt.source
def connector_source(api_key: str = dlt.secrets.value):
    config: RESTAPIConfig = {
        "client": {
            "base_url": "https://api.example.com/",
            "auth": {...},
        },
        "resources": [...],
    }
    yield from rest_api_resources(config)

def run_pipeline():
    pipeline = dlt.pipeline(
        pipeline_name=DLT_SOURCE_NAME,
        destination=DLT_DESTINATION,
        dataset_name=DLT_SOURCE_NAME,
    )
    load_info = pipeline.run(connector_source())
    print(load_info)

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

## Source Implementation Patterns

### Pattern 1: Simple REST API Source

For straightforward public APIs:

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

source = rest_api_source({
    "client": {"base_url": "https://api.example.com/"},
    "resource_defaults": {
        "endpoint": {"params": {"limit": 100}},
        "write_disposition": "replace",
    },
    "resources": ["users", "posts", "comments"],
})
```

### Pattern 2: REST API with Authentication

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

@dlt.source
def my_api_source(api_token: str = dlt.secrets.value):
    config: RESTAPIConfig = {
        "client": {
            "base_url": "https://api.example.com/",
            "auth": {
                "type": "api_key",
                "name": "Authorization",
                "api_key": f"Bearer {api_token}",
                "location": "header",
            },
        },
        "resource_defaults": {
            "primary_key": "id",
            "write_disposition": "merge",
        },
        "resources": [...],
    }
    yield from rest_api_resources(config)
```

### Pattern 3: Custom Source Module

For complex APIs requiring custom logic:

```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
```

## Data Transformation Pattern

For connectors with `get_cleaner_callables()` transformations:

```python
def extract_domain_from_email(record: dict) -> dict:
    email = record.get("email")
    if email and "@" in email:
        record["email_domain"] = email.split("@")[1]
    return record

# Apply in pipeline:
resource.add_map(extract_domain_from_email)
```

## secrets_example.toml Template

```toml
# Copy this file to secrets.toml and fill in your credentials

[runtime]
log_level = "WARNING"
dlthub_telemetry = false

[load]
delete_completed_jobs = true

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

[sources.<connector_name>]
api_key = "your_api_key"

[destination.postgres.credentials]
host = "localhost"
port = 5432
database = "your_database"
username = "your_username"
password = "your_password"
```