Run a dlt ingest pipeline
You want Python ingestion that handles API calls, schema drift, state, and load packages without hand-writing every INSERT. In this guide, a Flight runs a dlt pipeline that fetches public GitHub repository metadata, loads it into docs_playground.flights_demo_dlt.github_repo_stats, and records each run in docs_playground.flights_demo.dlt_ingest_runs.
The dlt dataset and run ledger live in your own MotherDuck account, so you can inspect the generated tables, replace the demo repo_rows() source with a real API, or add the output to a dashboard.
Before you start
The Flight runtime authenticates to MotherDuck for you and injects the credential as MOTHERDUCK_TOKEN, which dlt's MotherDuck destination picks up automatically. To run a scheduled Flight as a service account instead, see Authentication, config, and secrets.
Use dlt[motherduck] with destination="motherduck" and pass loader_file_format="parquet" in the run call. That keeps the Flight's loading path explicit and avoids falling back to row-wise insert_values if the destination or loader config changes.
The demo uses a small public GitHub API call so you can run the whole flow without extra credentials. If you replace it with a private API, keep secrets out of Flight config and read credentials from a MotherDuck-managed secret or another short-lived credential source.
Create the Flight
Create the Flight. The code sets the dlt MotherDuck destination database to docs_playground; the injected MOTHERDUCK_TOKEN supplies the credential. Create it on demand first, then add a schedule after you verify the run.
SELECT flight_id, flight_name, current_version
FROM MD_CREATE_FLIGHT(
name := 'docs_dlt_ingest',
requirements_txt := array_to_string([
'duckdb==1.5.3',
'dlt[motherduck]==1.27.0',
'httpx==0.28.1'
], chr(10)),
source_code := $flight$
import os
import duckdb
import dlt
import httpx
REPOS = [
"duckdb/duckdb",
"motherduckdb/motherduck-docs",
"dlt-hub/dlt",
]
def repo_rows():
for repo in REPOS:
response = httpx.get(
f"https://api.github.com/repos/{repo}",
timeout=30,
headers={"Accept": "application/vnd.github+json"},
)
response.raise_for_status()
payload = response.json()
yield {
"repo": repo,
"stars": payload.get("stargazers_count"),
"forks": payload.get("forks_count"),
"open_issues": payload.get("open_issues_count"),
"default_branch": payload.get("default_branch"),
"pushed_at": payload.get("pushed_at"),
"loaded_at": payload.get("updated_at"),
}
def main():
os.environ.setdefault("HOME", "/tmp")
os.environ["DESTINATION__MOTHERDUCK__CREDENTIALS__DATABASE"] = "docs_playground"
pipeline = dlt.pipeline(
pipeline_name="flights_github_repo_stats",
destination="motherduck",
dataset_name="flights_demo_dlt",
)
load_info = pipeline.run(
repo_rows(),
table_name="github_repo_stats",
write_disposition="merge",
primary_key="repo",
loader_file_format="parquet",
)
con = duckdb.connect("md:")
con.execute("CREATE SCHEMA IF NOT EXISTS docs_playground.flights_demo")
con.execute("""
CREATE TABLE IF NOT EXISTS docs_playground.flights_demo.dlt_ingest_runs (
run_at TIMESTAMPTZ,
pipeline_name VARCHAR,
destination_dataset VARCHAR,
load_summary VARCHAR
)
""")
con.execute(
"""
INSERT INTO docs_playground.flights_demo.dlt_ingest_runs
VALUES (current_timestamp, ?, ?, ?)
""",
["flights_github_repo_stats", "flights_demo_dlt", str(load_info)],
)
print(load_info)
if __name__ == "__main__":
main()
$flight$
);Run and inspect it
The MD_* Flight table functions only accept literal parameters, not subqueries or lateral join columns, so store the Flight ID in a SQL variable first. The next cells reuse it through getvariable:
SET VARIABLE dlt_flight_id = ( SELECT flight_id FROM MD_LIST_FLIGHTS() WHERE flight_name = 'docs_dlt_ingest' ORDER BY created_at DESC LIMIT 1 );
Trigger a manual run:
SELECT *
FROM MD_RUN_FLIGHT(
flight_id := getvariable('dlt_flight_id')
);Poll for completion:
SELECT run_number, status, flight_version, created_at
FROM MD_LIST_FLIGHT_RUNS(
flight_id := getvariable('dlt_flight_id')
)
ORDER BY run_number DESC
LIMIT 5;Schedule the pipeline
After the manual run succeeds, add a daily 07:15 UTC schedule. Schedule updates are metadata-only; they do not create a new Flight version.
CALL MD_UPDATE_FLIGHT(
flight_id := getvariable('dlt_flight_id'),
schedule_cron := '15 7 * * *'
);Query the table dlt created:
SELECT repo, stars, forks, open_issues, default_branch, pushed_at FROM docs_playground.flights_demo_dlt.github_repo_stats ORDER BY stars DESC;
The ledger table captures the dlt load package summary:
SELECT run_at, pipeline_name, destination_dataset, load_summary FROM docs_playground.flights_demo.dlt_ingest_runs ORDER BY run_at DESC LIMIT 5;
Why this dlt setup
The important default is the load format. For MotherDuck, prefer Parquet loader files over row-wise insert_values. The Flight example makes that choice explicit with loader_file_format="parquet" so larger sources stay on a bulk-loading path.
Use this dlt pattern when you want schema evolution, state tracking, merge behavior, or a source connector. If you already have clean Parquet files in S3, the S3 guide is simpler. If you only have a few hundred rows of control metadata, direct inserts are fine.
Adapt the pattern
- Replace
repo_rows()with a dlt source for your API, database, or file system. - Move run-specific values such as the repo list into the Flight's
config(for example a comma-separatedREPOSkey read withos.environ), then override them for a single run with theconfigargument ofMD_RUN_FLIGHTinstead of editing the source. - Keep
DESTINATION__MOTHERDUCK__CREDENTIALS__DATABASEpointed at the database where dlt should create datasets. - Use
write_disposition="merge"with aprimary_keyfor entity tables andappendfor event streams. - Keep
loader_file_format="parquet"unless you have measured a reason to change it. - Lower dlt load workers if a source or network path is unreliable. See the dlt MotherDuck destination docs.