# Create SQL Database Pipeline

> Create a dlt pipeline from a SQL database source (postgres, mysql, mssql, oracle, sqlite, or any SQLAlchemy-supported database). Use when the user wants to load tables from a relational database to a destination like DuckDB, BigQuery, or Snowflake. Not for REST APIs or file sources.

- Skill: `dlt-hub/create-sql-database-pipeline` (Agent Skill)
- Install (CLI): `npx skillmds@latest add dlt-hub/create-sql-database-pipeline`
- Raw SKILL.md: https://api.skillmd.com/api/skills/dlt-hub/create-sql-database-pipeline/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: dlt-hub (https://skillmd.com/u/dlt-hub)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/dlt-hub/create-sql-database-pipeline

---


# Create a SQL database dlt pipeline

Build the simplest working pipeline — one table, no incremental loading — to get data flowing fast.

**Docs:** https://dlthub.com/docs/dlt-ecosystem/verified-sources/sql_database

Parse `$ARGUMENTS`:
- `database` (required): description of the source database (e.g. "postgres on localhost", "Rfam MySQL", a connection URL, or just the DB type)
- `destination` (optional, default `duckdb`): where to load data

## Steps

### 1. Assess data volume

Ask the user: **how much data will be loaded?** (approximate row count or table size is enough)

Note the answer — it will be used in step 10 to recommend the right backend and `chunk_size`.

Key rules regardless of scale:
- **Always pass `table_names=`** to `sql_database()` — avoids reflecting the entire schema
- **Large tables need incremental loading** — full reload of tens of millions of rows on every run is almost never the right plan; flag this to the user before writing code

### 2. Snapshot current folder

Run `ls -la` to see the current state before scaffolding.

### 3. Run dlthub pipeline init

```
uv run dlthub --non-interactive pipeline init sql_database <destination>
```

Note: `--non-interactive` is a global flag on `dlthub` and may appear at any position in the command. Always pass it to prevent prompts that block execution.

If the command fails with `invalid choice: 'pipeline'`, the dlthub workspace is not initialized. Run `uv run dlthub init` and follow its instructions — most importantly run `uv sync` to pull required dependencies — then retry.

This creates:
- `sql_database_pipeline.py` — working example
- `.dlt/secrets.toml` — credentials template
- `.dlt/config.toml` — pipeline config
- `requirements.txt`, `.gitignore`

Run `ls -la` again to confirm what was created.

### 4. Research before writing code

Do these in parallel:

**Read essential dlt docs upfront:**
- SQL database source overview: `https://dlthub.com/docs/dlt-ecosystem/verified-sources/sql_database` (setup steps: `https://dlthub.com/docs/dlt-ecosystem/verified-sources/sql_database/setup.md`)
- Backend options and performance: `https://dlthub.com/docs/dlt-ecosystem/verified-sources/sql_database/configuration.md`
- Credentials setup: `https://dlthub.com/docs/general-usage/credentials/setup.md`

**Identify the driver:**

Different databases need different SQLAlchemy dialect + driver packages:

| Database | drivername | extra package |
|---|---|---|
| PostgreSQL | `postgresql+psycopg2` | `psycopg2-binary` |
| MySQL / MariaDB | `mysql+pymysql` | `pymysql` |
| MS SQL Server | `mssql+pyodbc` | `pyodbc` |
| Oracle | `oracle+cx_oracle` | `cx_Oracle` |
| SQLite | `sqlite` | _(built-in)_ |

Install the driver + sql_database extras if missing:
```
uv add "dlt[hub,sql_database]" <driver-package>
```

### 5. Read generated files

Read the following (do NOT read `secrets.toml`):
- `sql_database_pipeline.py` — read for scaffold patterns, then replace its contents with the real pipeline in step 6
- `.dlt/config.toml` — pipeline config structure

### 6. Write the pipeline

**Replace the scaffold** — write the real pipeline into the generated `sql_database_pipeline.py`, or rename the file to match the use case (e.g. `movies_pipeline.py`). Either way: do not create a second file alongside the scaffold. The scaffold has no further purpose once replaced.

**Choose one scenario based on how many tables to load:**

#### Scenario A — Single table: use `sql_table`

```python
import dlt
from dlt.sources.sql_database import sql_table


def main() -> None:
    pipeline = dlt.pipeline(
        pipeline_name="<name>",
        destination="<destination>",
        dataset_name="<name>",
        dev_mode=True,       # fresh timestamped dataset on each run
        progress="log",      # NOTE: progress goes on dlt.pipeline(), NOT on pipeline.run()
    )

    table = sql_table(
        table="<table_name>",
        chunk_size=500,
        # schema="<schema>",  # set if not the default schema
    )

    load_info = pipeline.run(table.add_limit(1), write_disposition="replace")
    print(load_info)


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

#### Scenario B — Multiple tables: use `sql_database`

```python
import dlt
from dlt.sources.sql_database import sql_database


def main() -> None:
    pipeline = dlt.pipeline(
        pipeline_name="<name>",
        destination="<destination>",
        dataset_name="<name>",
        dev_mode=True,
        progress="log",
    )

    source = sql_database(
        schema="<schema>",           # set if not the default schema
        table_names=["<t1>", "<t2>"],  # always pass table_names — avoids full-schema reflection
        chunk_size=500,
    )

    load_info = pipeline.run(source.add_limit(1), write_disposition="replace")
    print(load_info)


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

Key rules:
- `progress="log"` belongs on `dlt.pipeline()`, **not** on `pipeline.run()` — that parameter does not exist on `run()`
- `dev_mode=True` creates a new timestamped dataset on every run — keeps test runs non-destructive
- `.add_limit(1)` loads one chunk only — always use for first/test runs
- **Always pass `table_names=`** to `sql_database()` — omitting it reflects the entire schema, which is slow and loads unwanted tables
- Do not hardcode credentials in the script — they are auto-injected from `.dlt/secrets.toml`

### 7. Apply reflection level

Using the reflection level chosen in `find-source`, set it on the source:

```python
# Scenario A
table = sql_table(table="<table_name>", chunk_size=500, reflection_level="full")

# Scenario B
source = sql_database(table_names=[...], chunk_size=500, reflection_level="full")
```

- `minimal` — column names and nullability only; types inferred from data. Use when `full` causes casting errors.
- `full` — column names, nullability, and data types including decimal precision/scale. Default; works for most cases.
- `full_with_precision` — maximum detail including precision for text/binary. Use when the destination requires strict typing; may cause type-mismatch errors.

Ref: https://dlthub.com/docs/dlt-ecosystem/verified-sources/sql_database/advanced#column-reflection

### 8. Add transformation callbacks (if needed)

If the user needs to transform data before or during loading, introduce the right tool depending on when the transformation happens:

- **Filter rows at SQL level** — `query_adapter_callback`:
  ```python
  def query_adapter(query, table):
      if table.name == "orders":
          return query.where(table.c.status == "active")
      return query

  source = sql_database(query_adapter_callback=query_adapter)
  ```

- **Add or modify columns at schema level** — `table_adapter_callback` to append computed columns before extraction.

- **Transform rows after extraction** — `add_map` on a resource:
  ```python
  def pseudonymize(row):
      row["email"] = hash(row["email"])
      return row

  source.orders.add_map(pseudonymize)
  ```

Ref: https://dlthub.com/docs/dlt-ecosystem/verified-sources/sql_database/usage

If no transformation is needed, skip this step.

> **Boundary:** these callbacks are for *extraction-time* transforms — filtering, masking, or reshaping data before it hits the destination. For *post-load* modeling (Kimball dimensions, CDM, cross-source joins) hand off to the **transformations** toolkit instead.

### 9. Set up config and secrets

**Config** (non-secret values like schema name, table name):

Edit `.dlt/config.toml` directly:

```toml
# .dlt/config.toml
[sources.sql_database]
table_name = "<table>"
```

**Secrets** (credentials): **never** read or write `secrets.toml` directly.

Present this template to the user and ask them to fill it in. Use `secrets_update_fragment` MCP tool (or `dlthub ai secrets` CLI) to write the fragment — do not edit the file directly:

```toml
# .dlt/secrets.toml
[sources.sql_database.credentials]
drivername = "<dialect+driver>"        # e.g. mysql+pymysql, postgresql+psycopg2
host = "<host>"
port = <port>
username = "<username>"
password = "<password>"
database = "<database>"
```

A connection string is also accepted:
```toml
[sources.sql_database.credentials]
credentials = "<dialect+driver>://user:password@host:port/database"
```

**ALWAYS Get Feedback** before running for the first time. Show a summary of the files you changed or created, and confirm the user has filled in credentials.

### 10. Debug pipeline — first run

When user is ready, run:
```
uv run python <pipeline_script>.py
```

Expected output shows extract/normalize/load steps with row counts and timing from `progress="log"`.

**When errors occur, use `debug-pipeline`** to diagnose — do not add more complexity first.

Common first-run errors:
- `ConfigFieldMissingException` — a credentials field is missing or misnamed in secrets.toml
- `OperationalError` / `Can't connect` — wrong host/port/credentials or DB unreachable
- `ModuleNotFoundError: No module named 'sqlalchemy'` — run `uv add "dlt[hub,sql_database]"`
- `MissingDependencyException: numpy required` — the pyarrow backend also needs numpy: `uv add numpy`

### 11. Suggest backend after a successful test run

Using the data volume noted in step 1, recommend the right backend:

| Scale | Rows / Size | Recommended backend | chunk_size |
| --- | --- | --- | --- |
| Small | < 100k rows / < 500 MB | `sqlalchemy` (default) | any |
| Medium | 100k – 10M rows / 500 MB – 10 GB | `pyarrow` (needs `numpy`) | 5000 |
| Large | > 10M rows / > 10 GB | `connectorx` (MySQL/Postgres) or `pyarrow` | 50000 |

Backend options:
- **`sqlalchemy`** — safest, works with every destination, but slowest.
- **`pyarrow`** — 20–30x faster; preserves decimal/date types precisely. Also requires `numpy`.
- **`pandas`** — convenient for DataFrame workflows, but loses precision on decimal and date columns.
- **`connectorx`** — fastest overall (2x over pyarrow, Rust-based); uses its own connection string format, bypasses SQLAlchemy.

Ref: https://dlthub.com/docs/dlt-ecosystem/verified-sources/sql_database/configuration#configuring-the-backend

Apply the chosen backend:
```python
table = sql_table(table="<table_name>", chunk_size=500, backend="pyarrow")
```

Re-run the test to confirm the backend works before moving on.

> **Note:** When using `pyarrow`, `pandas`, or `connectorx` (normalization skipped), `apply_hints` works for incremental loading, write disposition, merge keys, etc. — but schema changes like `columns={...}` do not work. Use `table_adapter_callback` for column-level schema changes instead.

## Next steps

- **Test run succeeded, backend chosen** → use `adjust-table` to remove limits and add incremental loading for production
- **Pipeline errors or 0 rows** → use `debug-pipeline` to inspect traces and load packages
- **Want more tables** → use `add-table` to add resources to the pipeline
- **Ready to explore data** → hand over to **data-exploration** toolkit
- **Ready to deploy** → hand over to **dlthub-platform** toolkit