---
name: create-dlt-connector
description: Scaffold a new dlt connector from scratch with full boilerplate — pipeline, pyproject.toml, secrets template, .env. Use when the user wants to create a new connector, init a connector, scaffold boilerplate, start a fresh data source, or add a new API integration to the monorepo.
argument-hint: "[connector-name]"
disable-model-invocation: true
---

# Create a New dlt Connector

Create a new dlt connector that extracts data from an API source and loads it to a destination database. Connectors run via `uv` for dependency isolation.

## Context

This skill targets a **dlt-connectors monorepo** structure where each connector is a self-contained directory under `connectors/`. Connectors use `uv` for dependency management and `dlt` for the pipeline framework. Each connector has its own `pyproject.toml` and `uv.lock`.

See `references/connector-framework-specification.md` for the full framework specification and `references/style-guide.md` for coding conventions.

## Source type decision

Before scaffolding, determine which dlt source pattern fits:

| Scenario | Pattern | When to use |
|----------|---------|-------------|
| Standard REST API with JSON, pagination, auth | Declarative REST API (`rest_api_source`) | Most common — config-based, minimal code |
| Complex API with custom logic, SDKs, or non-REST | Custom Python (`@dlt.source` + `@dlt.resource`) | When REST API config can't express the logic |
| Multiple related endpoints, same base URL | Declarative REST API with multiple resources | One config dict, many endpoints |

See the `add-dlt-rest-api-source` skill for detailed REST API patterns (pagination types, auth, incremental loading).

## Prerequisites

Before starting, ensure clarity on:

- Connector name: `$ARGUMENTS` (lowercase, underscores for spaces)
- API documentation available for the data source
- Authentication method known (API key, OAuth, etc.)
- Which resources/endpoints to extract
- Write disposition for each resource (replace or merge)
- Primary keys for merge operations (if applicable)

**If any of the above are unclear, ask the user before proceeding.**

## Steps

1. Create the connector directory structure.

   ```bash
   mkdir -p connectors/<connector_name>/.dlt
   cd connectors/<connector_name>
   ```

2. Create `pyproject.toml` with dlt and any connector-specific dependencies.

   ```toml
   [project]
   name = "dlt-connector-<connector_name>"
   version = "0.1.0"
   requires-python = ">=3.12"
   dependencies = [
       "dlt[duckdb,postgres]>=1.14.1",
   ]

   [dependency-groups]
   dev = [
       "black>=25.1.0",
       "ipdb>=0.13.13",
   ]
   ```

3. Generate the lock file.

   ```bash
   uv lock
   ```

4. Create `<connector_name>_pipeline.py` with the required module-level environment variables and pipeline configuration.

   ```python
   import os
   import dlt

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


   def run_pipeline():
       pipeline = dlt.pipeline(
           pipeline_name=DLT_SOURCE_NAME,
           destination=DLT_DESTINATION,
           dataset_name=DLT_SOURCE_NAME,
           dev_mode=False,
       )

       source = your_source()  # Replace with actual source

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


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

5. Create `.env.example` with the standard environment variables.

   ```env
   DLT_DESTINATION=postgres
   DLT_SOURCE_NAME=<connector_name>
   ```

6. Create `.gitignore` to exclude secrets and artifacts.

   ```gitignore
   .dlt/*
   !.dlt/secrets_example.toml
   .env
   __pycache__/
   *.pyc
   .venv/
   *.duckdb
   *.duckdb.wal
   ```

7. Create `.dlt/secrets_example.toml` with required dlt settings and credential placeholders.

   ```toml
   # Copy this file to secrets.toml and fill in your credentials
   # DO NOT commit secrets.toml to git!

   [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_here"

   [destination.postgres.credentials]
   host = "localhost"
   port = 5434
   database = "postgres"
   username = "dlt"
   password = "dev12345_"
   ```

8. Implement the source using the appropriate pattern (see the `add-dlt-rest-api-source` skill for REST API patterns, or `references/connector-framework-specification.md` for custom source modules).

9. Test the pipeline locally.

   ```bash
   cp .dlt/secrets_example.toml .dlt/secrets.toml
   # Edit .dlt/secrets.toml with real credentials
   uv sync
   DLT_DESTINATION=duckdb uv run python <connector_name>_pipeline.py
   ```

10. Verify data was loaded successfully. If using Postgres:

    ```bash
    PGPASSWORD=dev12345_ psql -h localhost -p 5434 -U dlt -d postgres -c "
    SELECT table_name FROM information_schema.tables
    WHERE table_schema = '<connector_name>' ORDER BY table_name;"
    ```

11. Create `README.md` with overview, resources table, configuration instructions, and run commands.

## Validation

- [ ] `uv sync` succeeds
- [ ] Pipeline loads data locally (`uv run python <connector_name>_pipeline.py`)
- [ ] Data verified in destination (tables exist, rows loaded)
- [ ] `README.md` is complete
- [ ] All unclear requirements were clarified with user

## Examples

```bash
# Full workflow for a new "shopify" connector
mkdir -p connectors/shopify/.dlt
cd connectors/shopify
# ... create files per steps above ...
uv sync
DLT_DESTINATION=duckdb uv run python shopify_pipeline.py
```