data load tool (dlt) — the open-source Python library that automates all your tedious data loading tasks

Be it a Google Colab notebook, AWS Lambda function, an Airflow DAG, your local laptop,
or an AI coding agent—dlt can be dropped in anywhere.

🚀 Join our thriving community of likeminded developers and build the future together!

Installation

dlt supports Python 3.10 through Python 3.14. Note that some optional extras are not yet available for Python 3.14, so support for this version is considered experimental.

pip install dlt

Add the extras you need for your sources and destinations, for example:

pip install "dlt[duckdb]"        # local DuckDB destination
pip install "dlt[bigquery]"      # or snowflake, postgres, redshift, databricks, athena, ...
pip install "dlt[s3]"            # or gs, az for cloud filesystems
pip install "dlt[sql_database]"  # read from any SQL database
pip install "dlt[hub]"           # data quality, transformations, and AI (see below)

Prefer uv? uv add "dlt[duckdb]".

Quick Start

Describe an API declaratively and load it into DuckDB — dlt handles requests, pagination, schema inference, and typing for you:

import dlt
from dlt.sources.rest_api import rest_api_source

# 1. Describe the API declaratively
source = rest_api_source({
    "client": {"base_url": "https://api.spotify.com/v1"},
    "resources": [
        {
          "name": "playlist_tracks",
          "endpoint": {"path": "playlists/{playlist_id}/tracks"},
        },
    ],
})

# 2. Point a pipeline at any destination
pipeline = dlt.pipeline(
    pipeline_name="spotify",
    destination="duckdb",
    dataset_name="spotify_data",
)

# 3. Extract, normalize, and load
pipeline.run(source)

# 4. ...and read it straight back as a DataFrame
pipeline.dataset().playlist_tracks.df()

...or load any Python iterable — a resource is just a generator, and dlt infers the schema, types the columns, and writes the table:

import dlt

@dlt.resource(table_name="tracks", primary_key="id", write_disposition="merge")
def tracks():
    yield {"id": 1, "title": "Yellow",       "artist": "Coldplay",   "streams": 4_200_000_000}
    yield {"id": 2, "title": "Shape of You", "artist": "Ed Sheeran", "streams": 3_900_000_000}

dlt.pipeline(
    destination="duckdb",
    dataset_name="spotify_data",
).run(
  source=tracks(),
)

Check out a basic in Colab or a more advanced Hugging Face demo with Marimo notebooks.

Why dlt

dlt loads data from messy, often unstructured sources into well-structured, typed datasets. It's a library, not a platform — you pip install it into your existing code and keep your workflow and the other tools you already use. No black boxes: clean Pythonic interfaces, human-readable file formats, schemas you can inspect, no hidden side effects.

dlt and its docs are built from the ground up for LLMs and coding agents. Pair the typed, declarative primitives below with dlthub.com/context and the LLM-native workflow to go from prompt to working pipeline — across 5000+ sources — often in a single shot.

Extract from any source

REST APIs — describe the endpoints declaratively; filter, map, and flatten records right at the source (docs):

from dlt.sources.rest_api import rest_api_source

source = rest_api_source({
    "client": {
        "base_url": "https://api.spotify.com/v1",
        "paginator": {"type": "cursor", "cursor_path": "next_cursor"},
    },
    "resources": [
        {
            "name": "playlist_tracks",
            "endpoint": {"path": "playlists/{playlist_id}/tracks"},
            "processing_steps": [
                {"filter": lambda r: r["track"]["duration_ms"] > 0},
                {"map": flatten_track},
            ],
        },
    ],
})

def flatten_track(record: dict[str, Any]) -> dict[str, Any]:
    ...

SQL databases — reflect tables and types straight from the database (docs):

from dlt.sources.sql_database import sql_database

source = sql_database("mysql+pymysql://user:pass@host/spotify")

Files in any bucket — list, then parse CSV / JSONL / Parquet from local disk, S3, GCS, or Azure (docs):

from dlt.sources.filesystem import filesystem, read_csv_duckdb

source = (
    filesystem(
        bucket_url="s3://my-bucket/spotify",
        file_glob="tracks_*.csv",
    ) | read_csv_duckdb()
).with_name("tracks")

DataFrames & Arrow — pandas, Polars, and Arrow tables load directly; Arrow-backed frames move with zero copies:

import dlt
import pandas as pd

df = pd.DataFrame({
  "track": ["Yellow",        "Shape of You"],
  "streams": [4_200_000_000, 3_900_000_000],
})
dlt.pipeline(
    destination="duckdb",
    dataset_name="spotify_data",
).run(
    df,
    table_name="tracks",
)

See many more sources in the ecosystem.

Load to 20+ destinations — swap one string

The same resource runs anywhere. Change the destination string and dlt takes care of credentials, DDL in the target dialect, staging, and schema drift:

pipeline = dlt.pipeline(
    pipeline_name="spotify",
    destination="duckdb",         # → snowflake, bigquery, postgres, redshift, databricks,
    dataset_name="spotify_data",  #   athena, clickhouse, motherduck, filesystem (S3/GCS/Azure),
)                                 #   iceberg, delta, ... and custom reverse-ETL destinations
pipeline.run(source)

dlt handles the parts you'd rather not:

• Credentials → secrets.toml / env vars, injected automatically
• DDL → CREATE TABLE in the target's dialect
• Type mapping → source types converted to the destination's types
• Staging → S3 / GCS for warehouses that need it
• Schema drift → ALTER TABLE on the fly

Browse all supported destinations, or build a custom one.

Declare intent with decorators

Decorators let you declare what you want — incremental loading, merge strategies, schema contracts, column hints — instead of hand-rolling it. Every knob can be overridden at runtime (docs):

import dlt

@dlt.resource(
    primary_key="id",
    write_disposition="merge",                        # upsert on the primary key
    columns={"artist": {"x-annotation-pii": False}},  # type and annotate columns
    schema_contract={"columns": "freeze"},            # reject unexpected columns
)
def tracks(
    updated_at=dlt.sources.incremental("updated_at"),  # load only new/changed rows
):
    yield from fetch_tracks(since=updated_at.last_value)


@dlt.source
def spotify(api_key: str = dlt.secrets.value):
    return tracks(), playlists()   # group one or more resources behind shared config/auth

Schema contracts enforce the shape at the gate, with three modes — evolve (accept and adapt the schema), freeze (reject the record), and discard (drop the offending row/column) — applied independently to tables, columns, and data_type. You also get schema inference, normalization of nested data, incremental loading, and [secrets & config injection](