Motherduck Create Flight

Create and Manage MotherDuck Flights

Use this skill when the user needs Python to run on MotherDuck — on a schedule or on demand — instead of in their own infrastructure. A Flight is a single-file Python program that MotherDuck executes in a managed runtime with a MotherDuck token injected, pip dependencies installed from a requirements.txt, and stdout/stderr captured as run logs. The primary use cases are scheduled ingestion (pull from Postgres, S3, APIs, other warehouses, or any dlt source into MotherDuck tables) and scheduled transformation (refresh aggregates, run dbt, recompute reporting tables).

Source Of Truth

• Non-negotiable ordering: when MotherDuck MCP is available, call get_flight_guide before create_flight, update_flight, or edit_flight_source. The guide defines the current authoring contract, runtime limits, and tool semantics.
• Prefer current MotherDuck Flights docs over memory; the feature is in Preview and details shift.
• Without MCP, the same operations exist as SQL functions (MD_CREATE_FLIGHT, MD_RUN_FLIGHT, MD_FLIGHTS(), ...) that execute server-side on a MotherDuck connection. Parameter names differ slightly between the two surfaces; see the naming table in references/FLIGHTS_GUIDE.md.

Default Posture

• One Flight = one single-file Python script with def main(): ... and if __name__ == "__main__": main(). No CLI args — every knob comes from env vars via config (non-secret) or TYPE flights secrets (sensitive).
• Connect with duckdb.connect("md:"); the runtime injects MOTHERDUCK_TOKEN automatically. Never hardcode a token in source, config, or requirements.
• Always pin dependencies in requirements_txt, and pin duckdb to the latest MotherDuck-supported version (currently duckdb==1.5.2); an unpinned duckdb can install a release MotherDuck does not accept yet and fail at connect.
• Each secret param arrives as the env var <secret_name>_<PARAM>, not the bare param name. This is the single most common authoring mistake.
• Bulk-load, never row-by-row: stage to /tmp/ and read_csv_auto/read_json_auto/read_parquet, or one CTAS / INSERT ... SELECT. No executemany() against MotherDuck.
• Make every run idempotent: CREATE OR REPLACE TABLE full refresh, partition DELETE + INSERT, or dlt write_disposition="merge" with a primary key. Bootstrap with CREATE DATABASE IF NOT EXISTS / CREATE SCHEMA IF NOT EXISTS so the first run succeeds on a fresh account.
• Validate any config-supplied identifier (database, schema, table names) against [A-Za-z_][A-Za-z0-9_]* before interpolating it into DDL; bind all data values as ? parameters.
• Create the flight without a schedule first, trigger one on-demand run, read the logs, and only then attach schedule_cron (5-field cron, UTC).
• For production, use a service-account token via access_token_name and keep its database permissions as narrow as the workload allows.
• A Flight is sized for orchestration and light processing (2 cores, 16GB RAM, ~150GB scratch at /tmp/), not for crunching large tables in Python memory — push heavy compute into SQL.

Workflow

1. Classify the job: ingestion, transformation/refresh, export or alerting, or admin automation. If the job is interactive analysis or a one-off query, use motherduck-query instead — no Flight needed. 2. Call get_flight_guide (MCP) and confirm which database the flight writes to with motherduck-explore. 3. Start from the closest template in references/FLIGHT_EXAMPLES.md (dlt source, Postgres mirror, or S3 partition refresh) rather than writing from scratch; adapt via config knobs, not code surgery. 4. Create any required TYPE flights secret first, then create_flight with name, source_code, pinned requirements_txt, config, and secret names — no schedule_cron yet. 5. run_flight, poll list_flight_runs until terminal, and read get_flight_logs. Iterate with edit_flight_source (surgical) or update_flight (full field replacement); each content change creates a new version. 6. Once a run succeeds, set the schedule with update_flight(schedule_cron = ...) and tell the user the cron is UTC. Clear it later with schedule_cron = "".

Open Next

• Read references/FLIGHTS_GUIDE.md for the full concept and operations reference: anatomy, runtime environment, config vs secrets, scheduling, versioning, run lifecycle, the complete MCP tool reference, MCP-vs-SQL naming, loading strategies by data volume, and troubleshooting.
• Read references/FLIGHT_EXAMPLES.md for three complete, best-practice flight templates (dlt ingestion, Postgres ingestion, scheduled S3 partition refresh) with their requirements.txt, secret setup, and deploy calls.

Related Skills

• motherduck-load-data for choosing the ingestion SQL the flight will run (CTAS, INSERT ... SELECT, cloud-storage secrets)
• motherduck-query for validating the DuckDB SQL inside the flight before deploying it
• motherduck-explore for confirming target databases, schemas, and tables exist
• motherduck-build-data-pipeline when the work is a full raw/staging/analytics pipeline design and the flight is just its scheduler

Flight Examples

Three complete, best-practice flight templates adapted from MotherDuck's flight-plans templates: a dlt API ingestion flight, a Postgres mirroring flight, and a scheduled S3 partition refresh. Start from the closest one and adapt it through config knobs rather than rewriting it.

Contents

Shared Conventions

Every template follows the same contract; preserve these when adapting:

• def main() -> None: entrypoint with if __name__ == "__main__": main(); single file; no CLI args.
• All knobs from env vars through an env(name, default) helper, so users adapt by setting config values, not editing code.
• duckdb.connect("md:") with the injected MOTHERDUCK_TOKEN; no credentials in source.
• IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") validation for every config-supplied name interpolated into DDL; ? parameters for all data values.
• CREATE DATABASE IF NOT EXISTS / CREATE SCHEMA IF NOT EXISTS bootstrap so the first run works on a fresh account.
• Idempotent writes (merge, atomic CREATE OR REPLACE, or partition DELETE + INSERT) plus an append-only run ledger with run_at TIMESTAMPTZ first.
• print()/logging to stdout (captured as run logs); exit non-zero on failure.
• Pinned requirements_txt with duckdb==1.5.2 (latest MotherDuck-supported release).

Example: dlt Ingestion Flight

Runs a dlt pipeline into MotherDuck on a schedule: Parquet loader files, schema evolution, merge semantics keyed on PRIMARY_KEY, and a run ledger. The demo source is public GitHub repo metadata (no credentials) so a fresh deploy produces a successful run; replace repo_rows with any dlt source — an API, a database, a filesystem, or a dlt verified source.

import os
import re
from collections.abc import Iterator

import dlt
import duckdb
import httpx


IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
WRITE_DISPOSITIONS = {"append", "merge", "replace"}


def repo_rows(repos: list[str]) -> Iterator[dict]:
    # Demo source: public GitHub repository metadata, no credentials needed.
    # Replace this generator with your own dlt source (an API, a database, a
    # filesystem, or a dlt verified source) to ingest real data. Yield plain
    # dicts and dlt infers the schema and evolves it as fields change.
    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"),
        }


def main() -> None:
    # Every knob is read from Flight config/env, so you adapt this template by
    # setting config values rather than editing code. Defaults load public GitHub
    # repo stats into flights_demo so a fresh deploy produces a successful run.
    database = validate_identifier("DESTINATION_DATABASE", env("DESTINATION_DATABASE", "flights_demo"))
    dataset_name = env("DATASET_NAME", "flights_demo_dlt")
    table_name = env("TABLE_NAME", "github_repo_stats")
    pipeline_name = env("PIPELINE_NAME", "flights_dlt_ingest")
    primary_key = env("PRIMARY_KEY", "repo")
    write_disposition = env("WRITE_DISPOSITION", "merge")
    if write_disposition not in WRITE_DISPOSITIONS:
        raise ValueError(
            f"WRITE_DISPOSITION must be one of {sorted(WRITE_DISPOSITIONS)}, got {write_disposition!r}"
        )
    ledger_table = validate_identifier("RUN_LEDGER_TABLE", env("RUN_LEDGER_TABLE", "dlt_ingest_runs"))
    repos = [
        repo.strip()
        for repo in env("GITHUB_REPOS", "duckdb/duckdb,motherduckdb/motherduck-docs,dlt-hub/dlt").split(",")
        if repo.strip()
    ]

    # dlt writes working files under HOME; a Flight has a writable /tmp.
    os.environ.setdefault("HOME", "/tmp")
    # Point the dlt MotherDuck destination at our database. The injected
    # MOTHERDUCK_TOKEN supplies the credential, so no token appears here.
    os.environ["DESTINATION__MOTHERDUCK__CREDENTIALS__DATABASE"] = database

    # Create the destination database so dlt has a catalog to build the dataset in;
    # dlt creates the dataset (schema) and tables, but not the database itself.
    con = duckdb.connect("md:")
    con.execute(f"CREATE DATABASE IF NOT EXISTS {database}")

    pipeline = dlt.pipeline(
        pipeline_name=pipeline_name,
        destination="motherduck",
        dataset_name=dataset_name,
    )
    load_info = pipeline.run(
        repo_rows(repos),
        table_name=table_name,
        write_disposition=write_disposition,
        primary_key=primary_key,
        # Prefer Parquet loader files over row-wise insert_values so larger
        # sources stay on a bulk-loading path. Keep this unless you have measured
        # a reason to change it.
        loader_file_format="parquet",
    )

    # Record the dlt load package summary so each run leaves an audit trail. The
    # ledger lives in the database's main schema, separate from the dlt dataset.
    con.execute(f"CREATE SCHEMA IF NOT EXISTS {database}.main")
    con.execute(
        f"""
        CREATE TABLE IF NOT EXISTS {database}.main.{ledger_table} (
            run_at TIMESTAMPTZ,
            pipeline_name VARCHAR,
            destination_dataset VARCHAR,
            destination_table VARCHAR,
            load_summary VARCHAR
        )
        """
    )
    con.execute(
        f"INSERT INTO {database}.main.{ledger_table} VALUES (current_timestamp, ?, ?, ?, ?)",
        [pipeline_name, dataset_name, table_name, str(load_info)],
    )
    con.close()
    print(load_info)


def env(name: str, default: str) -> str:
    value = os.environ.get(name, default).strip()
    return value or default


def validate_identifier(name: str, value: str) -> str:
    # The database and ledger table names flow into CREATE/INSERT statements that
    # cannot be parameterized, so reject anything that is not a plain SQL
    # identifier before any SQL runs.
    if not IDENTIFIER_RE.fullmatch(value):
        raise ValueError(f"{name} must be a simple SQL identifier, got {value!r}")
    return value


if __name__ == "__main__":
    main()

requirements_txt:

duckdb==1.5.2
dlt[motherduck]==1.27.0
httpx==0.28.1

Config knobs (all optional): DESTINATION_DATABASE, DATASET_NAME, TABLE_NAME, PIPELINE_NAME, PRIMARY_KEY, WRITE_DISPOSITION (append/merge/replace), RUN_LEDGER_TABLE, GITHUB_REPOS. When you swap in a credentialed source, put the credentials in a TYPE flights secret and read them as <secret_name>_<PARAM> env vars.

Example: Postgres Ingestion Flight

Mirrors PostgreSQL base tables into a MotherDuck database using the DuckDB postgres core extension. Each table moves in one streaming statement — CREATE OR REPLACE TABLE <target> AS SELECT * FROM pg."schema"."table" — which is atomic (one-step swap), idempotent (rerun fully replaces), and memory-bounded (DuckDB pipelines the scan into the write). Includes/excludes are config-driven, each table gets jittered exponential-backoff retries with per-table failure isolation, and results land in an audit table.

Create the connection secret first (params must be UPPERCASE; they arrive as pg_HOST, pg_PASSWORD, ... because the unquoted secret name is lowercased):

CREATE SECRET pg IN motherduck (
    TYPE flights,
    PARAMS MAP {
        'HOST':     '<your-postgres-host>',
        'PORT':     '5432',
        'DATABASE': '<your_database>',
        'USER':     '<YOUR_USER>',
        'PASSWORD': '<YOUR_PASSWORD>',
        'SSLMODE':  'require'
    }
);

"""
Postgres -> MotherDuck batch ELT flight

Mirrors PostgreSQL base tables into a MotherDuck database using the DuckDB postgres
extension. Each table is moved by a single streaming SQL statement that is atomic,
idempotent, and memory-bounded. Per-table logging lands in
<target>.main.flight_tracker.

Inputs (case sensitive, use uppercase):
    Secret `pg` (TYPE flights) -> Postgres connection params:
        Required: HOST, DATABASE, USER, PASSWORD
        Optional: PORT, SSLMODE
    Config (non-secret env vars):
        MOTHERDUCK_HOST  - optional host override; exported before connect.
        TARGET_DATABASE  - MotherDuck database to write into (default: postgres_ingest).
        INCLUDED_SCHEMAS / EXCLUDED_SCHEMAS  - comma-separated schema names.
        INCLUDED_TABLES  / EXCLUDED_TABLES   - comma-separated, fully qualified schema.table.
        MAX_RETRIES (5)
        RETRY_BASE_SECONDS (2)
"""

from __future__ import annotations

import logging
import os
import sys
import uuid
from datetime import datetime, timezone

import duckdb
from tenacity import (
    Retrying,
    stop_after_attempt,
    wait_exponential,
    wait_random,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    stream=sys.stdout,
)
log = logging.getLogger("pg2md")

# PostgreSQL system schemas that are never mirrored.
SYSTEM_SCHEMAS = {"information_schema", "pg_catalog", "pg_toast"}

# Postgres connection params from the flight secret.
# The secret injects each as `<secret_name>_<KEY>`
PG_PARAMS = (
    ("HOST", "PGHOST", None, True),
    ("PORT", "PGPORT", "5432", False),
    ("DATABASE", "PGDATABASE", None, True),
    ("USER", "PGUSER", None, True),
    ("PASSWORD", "PGPASSWORD", None, True),
    ("SSLMODE", "PGSSLMODE", "prefer", False),
)

# Local DuckDB catalog name the source Postgres database is ATTACHed as. Referenced by
# attach_postgres(), discover_base_tables(), and load_table() -- one source of truth.
PG_ALIAS = "pg"


# --------------------------------------------------------------------------- #
# Small SQL / env helpers
# --------------------------------------------------------------------------- #
def quote_ident(ident: str) -> str:
    """Quote an identifier the way DuckDB/Postgres expect, so names with special
    characters or reserved words are handled correctly."""
    return '"' + ident.replace('"', '""') + '"'


def csv_set(name: str) -> frozenset[str]:
    """Turn a comma-separated env var into a clean set for membership filtering."""
    raw = os.environ.get(name, "") or ""
    return frozenset(part.strip() for part in raw.split(",") if part.strip())


# --------------------------------------------------------------------------- #
# Table selection
# --------------------------------------------------------------------------- #
def is_selected(
    schema: str,
    table: str,
    included_schemas: frozenset[str],
    excluded_schemas: frozenset[str],
    included_tables: frozenset[str],
    excluded_tables: frozenset[str],
) -> bool:
    """Decide whether a discovered base table is mirrored, applying the two
    include/exclude gates where exclude always wins and system schemas are excluded."""
    fqtn = f"{schema}.{table}"
    if schema in SYSTEM_SCHEMAS or schema.startswith("pg_temp") or schema.startswith("pg_toast"):
        return False
    if included_schemas and schema not in included_schemas:
        return False
    if schema in excluded_schemas:
        return False
    if included_tables and fqtn not in included_tables:
        return False
    if fqtn in excluded_tables:
        return False
    return True


# --------------------------------------------------------------------------- #
# Connection + setup
# --------------------------------------------------------------------------- #
def connect_motherduck() -> duckdb.DuckDBPyConnection:
    """Open the MotherDuck connection that backs the whole flight, targeting the
    configured host when one is set."""
    host = os.environ.get("MOTHERDUCK_HOST")
    if host:
        os.environ["motherduck_host"] = host
        log.info("Targeting MotherDuck host: %s", host)
    else:
        log.info("MOTHERDUCK_HOST not set; using runtime default MotherDuck host")
    return duckdb.connect("md:")


def attach_postgres(con: duckdb.DuckDBPyConnection, secret_name: str) -> None:
    """Wire up the read-only Postgres source so tables can be streamed out, keeping the
    password out of SQL by passing credentials through libpq env vars.
    ATTACHes READ_ONLY as `pg`."""
    for key, libpq_var, default, required in PG_PARAMS:
        env_var = f"{secret_name}_{key}"
        value = os.environ.get(env_var, default)
        if value is None:
            if required:
                raise RuntimeError(f"Required Postgres secret env var {env_var!r} is not set")
            continue
        os.environ[libpq_var] = str(value)

    con.execute("INSTALL postgres")
    con.execute("LOAD postgres")
    con.execute(f"ATTACH '' AS {PG_ALIAS} (TYPE postgres, READ_ONLY)")
    log.info(
        "Attached Postgres %s:%s/%s (read-only, sslmode=%s)",
        os.environ["PGHOST"], os.environ["PGPORT"],
        os.environ["PGDATABASE"], os.environ["PGSSLMODE"],
    )


def ensure_target(con: duckdb.DuckDBPyConnection, target_db: str) -> None:
    """Create the target database and the audit logging table up front."""
    target = quote_ident(target_db)
    con.execute(f"CREATE DATABASE IF NOT EXISTS {target}")
    con.execute(
        f"CREATE TABLE IF NOT EXISTS {target}.main.flight_tracker ("
        "  run_id               VARCHAR,"
        "  flight_secret_name   VARCHAR,"
        "  source_schema        VARCHAR,"
        "  source_table         VARCHAR,"
        "  destination_database VARCHAR,"
        "  destination_schema   VARCHAR,"
        "  destination_table    VARCHAR,"
        "  rows_loaded          BIGINT,"
        "  attempts             INTEGER,"
        "  started_at           TIMESTAMP,"
        "  finished_at          TIMESTAMP,"
        "  update_ts            TIMESTAMP"
        ")"
    )


# --------------------------------------------------------------------------- #
# Discovery + per-table load
# --------------------------------------------------------------------------- #
def discover_base_tables(con: duckdb.DuckDBPyConnection) -> list[tuple[str, str]]:
    """List the candidate source tables to iterate on"""
    rows = con.execute(
        f"SELECT table_schema, table_name FROM postgres_query('{PG_ALIAS}', "
        "'SELECT table_schema, table_name FROM information_schema.tables "
        "WHERE table_type = ''BASE TABLE''') "
        "ORDER BY table_schema, table_name"
    ).fetchall()
    return [(r[0], r[1]) for r in rows]


def load_table(con: duckdb.DuckDBPyConnection, target_db: str, schema: str, table: str) -> int:
    """Perform the entire data movement for one table as a single atomic, idempotent,
    streaming CTAS. Returns the row count the CTAS reports as inserted."""
    tgt_table = f"{quote_ident(target_db)}.{quote_ident(schema)}.{quote_ident(table)}"
    src_table = f"{PG_ALIAS}.{quote_ident(schema)}.{quote_ident(table)}"
    return con.execute(f"CREATE OR REPLACE TABLE {tgt_table} AS SELECT * FROM {src_table}").fetchone()[0]


def record_success(
    con: duckdb.DuckDBPyConnection, target_db: str, run_id: str, secret_name: str,
    schema: str, table: str, rows_loaded: int, attempts: int,
    started_at: datetime, finished_at: datetime, update_ts: datetime,
) -> None:
    """After success, append a row to the audit table"""
    con.execute(
        f"INSERT INTO {quote_ident(target_db)}.main.flight_tracker "
        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        [run_id, secret_name, schema, table, target_db, schema, table,
         rows_loaded, attempts, started_at, finished_at, update_ts],
    )


# --------------------------------------------------------------------------- #
# main
# --------------------------------------------------------------------------- #
def main() -> None:
    """Orchestrate the full-refresh ELT: connect, attach, discover, then load each
    table sequentially with per-table retries/isolation and record results."""
    # Run config, read once from the environment and referenced as needed below.
    RUN_ID = str(uuid.uuid4())
    TARGET_DB = os.environ.get("TARGET_DATABASE", "postgres_ingest")
    MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "5"))
    RETRY_BASE_SECONDS = float(os.environ.get("RETRY_BASE_SECONDS", "2"))
    INCLUDED_SCHEMAS = csv_set("INCLUDED_SCHEMAS")
    EXCLUDED_SCHEMAS = csv_set("EXCLUDED_SCHEMAS")
    INCLUDED_TABLES = csv_set("INCLUDED_TABLES")
    EXCLUDED_TABLES = csv_set("EXCLUDED_TABLES")
    # MotherDuck Flights secret holding the Postgres connection; its params arrive as
    # <SECRET_NAME>_HOST, <SECRET_NAME>_PORT, ... Change it to point at another secret.
    SECRET_NAME = "pg"

    log.info("Run %s -> target %r", RUN_ID, TARGET_DB)

    con = connect_motherduck()
    attach_postgres(con, SECRET_NAME)
    ensure_target(con, TARGET_DB)

    all_tables = discover_base_tables(con)
    selected = [
        (s, t) for (s, t) in all_tables
        if is_selected(s, t, INCLUDED_SCHEMAS, EXCLUDED_SCHEMAS, INCLUDED_TABLES, EXCLUDED_TABLES)
    ]
    log.info("Discovered %d base table(s); %d selected after filters", len(all_tables), len(selected))

    if not selected:
        log.warning("No tables selected - nothing to do.")
        return

    # Pre-create the target schemas (mirroring source schema names) once.
    for sch in sorted({s for (s, _) in selected}):
        con.execute(f"CREATE SCHEMA IF NOT EXISTS {quote_ident(TARGET_DB)}.{quote_ident(sch)}")

    started_all = datetime.now(timezone.utc)
    failed: list[str] = []
    succeeded = 0
    rows_total = 0

    for schema, table in selected:
        fqtn = f"{schema}.{table}"
        started = datetime.now(timezone.utc)
        retryer = Retrying(
            stop=stop_after_attempt(MAX_RETRIES),
            wait=wait_exponential(multiplier=RETRY_BASE_SECONDS, max=60) + wait_random(0, 1),
            reraise=True,
        )
        try:
            rows = retryer(load_table, con, TARGET_DB, schema, table)
            attempts = retryer.statistics.get("attempt_number", 1)
            finished = datetime.now(timezone.utc)
            record_success(con, TARGET_DB, RUN_ID, SECRET_NAME, schema, table, rows,
                           attempts, started, finished, datetime.now(timezone.utc))
            succeeded += 1
            rows_total += rows
            log.info("OK   %-50s %12d rows (attempts=%d)", fqtn, rows, attempts)
        except Exception as exc:  # noqa: BLE001 - per-table isolation is intentional
            attempts = retryer.statistics.get("attempt_number", 1)
            failed.append(fqtn)
            log.error("FAIL %-50s (attempts=%d) %s: %s", fqtn, attempts, type(exc).__name__, exc)

    total_seconds = (datetime.now(timezone.utc) - started_all).total_seconds()
    log.info("Summary: %d succeeded, %d failed, %d rows in %.1fs (run %s)",
             succeeded, len(failed), rows_total, total_seconds, RUN_ID)

    if failed:
        log.error("Failed tables: %s", ", ".join(failed))
        sys.exit(1)


if __name__ == "__main__":
    main()

requirements_txt — the postgres extension is a DuckDB core extension loaded at runtime (INSTALL postgres; LOAD postgres), not a pip package, so there is no pip Postgres client dependency:

duckdb==1.5.2
tenacity==9.0.0

Example: Scheduled S3 Partition Refresh

Refreshes exactly one Hive partition of a MotherDuck table from partitioned Parquet in S3 on each run: schema inferred once via a zero-row CTAS, then DELETE + INSERT scoped to the partition so DuckDB prunes every other partition folder. Defaults read the public DuckDB PyPI download stats, partitioned by year. Override LOAD_PARTITION per run (via run_flight config overrides) to backfill specific partitions.

import os
import re
from datetime import datetime, timezone

import duckdb


IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")


def main() -> None:
    # Every knob is read from Flight config/env, so you adapt this template by
    # setting config values rather than editing code. Defaults point at the
    # public DuckDB PyPI download stats, partitioned in S3 by year.
    source_glob = env(
        "SOURCE_GLOB",
        "s3://us-prd-motherduck-open-datasets/pypi/duckdb/pypi_daily_stats/**/*.parquet",
    )
    partition_column = validate_identifier("PARTITION_COLUMN", env("PARTITION_COLUMN", "year"))
    database = validate_identifier("DESTINATION_DATABASE", env("DESTINATION_DATABASE", "flights_demo"))
    schema = validate_identifier("DESTINATION_SCHEMA", env("DESTINATION_SCHEMA", "main"))
    table = validate_identifier("DESTINATION_TABLE", env("DESTINATION_TABLE", "duckdb_pypi_downloads"))
    ledger_table = validate_identifier("RUN_LEDGER_TABLE", env("RUN_LEDGER_TABLE", "ingest_runs"))
    hive_partitioning = "true" if env_bool("HIVE_PARTITIONING", True) else "false"
    load_partition = resolve_partition(env("LOAD_PARTITION", ""))

    destination = f"{database}.{schema}.{table}"
    ledger = f"{database}.{schema}.{ledger_table}"

    con = duckdb.connect("md:")

    # The Flight creates its own destination, so it runs on the first deploy
    # without depending on a database or schema that already exists.
    con.execute(f"CREATE DATABASE IF NOT EXISTS {database}")
    con.execute(f"CREATE SCHEMA IF NOT EXISTS {database}.{schema}")

    # Create the destination once by inferring its columns from the source.
    # LIMIT 0 reads no rows, so this is cheap and keeps the destination's types
    # aligned with the source Parquet (including the partition column).
    con.execute(
        f"""
        CREATE TABLE IF NOT EXISTS {destination} AS
        SELECT *
        FROM read_parquet(?, hive_partitioning = {hive_partitioning})
        WHERE {partition_column} = ?
        LIMIT 0
        """,
        [source_glob, load_partition],
    )

    # Replace exactly one partition. Filtering on the partition column lets DuckDB
    # prune every other partition folder, so the scan cost stays flat as more
    # partitions land. To transform instead of copying through, replace this
    # SELECT * with your own projection or aggregation (keep the partition column).
    con.execute(f"DELETE FROM {destination} WHERE {partition_column} = ?", [load_partition])
    con.execute(
        f"""
        INSERT INTO {destination}
        SELECT *
        FROM read_parquet(?, hive_partitioning = {hive_partitioning})
        WHERE {partition_column} = ?
        """,
        [source_glob, load_partition],
    )

    row_count = con.execute(
        f"SELECT count(*) FROM {destination} WHERE {partition_column} = ?",
        [load_partition],
    ).fetchone()[0]

    # A lightweight audit trail of which partition each run refreshed.
    con.execute(
        f"""
        CREATE TABLE IF NOT EXISTS {ledger} (
            run_at TIMESTAMPTZ,
            source_glob VARCHAR,
            destination_table VARCHAR,
            partition_column VARCHAR,
            load_partition VARCHAR,
            row_count BIGINT
        )
        """
    )
    con.execute(
        f"INSERT INTO {ledger} VALUES (current_timestamp, ?, ?, ?, ?, ?)",
        [source_glob, destination, partition_column, str(load_partition), row_count],
    )
    print(f"refreshed {destination} partition {partition_column}={load_partition}: {row_count} rows")


def env(name: str, default: str) -> str:
    value = os.environ.get(name, default).strip()
    return value or default


def env_bool(name: str, default: bool) -> bool:
    value = os.environ.get(name)
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "t", "yes", "y", "on"}


def validate_identifier(name: str, value: str) -> str:
    # Database, schema, table, and column names flow into CREATE/DELETE/INSERT
    # statements that cannot be parameterized, so reject anything that is not a
    # plain SQL identifier before any SQL runs.
    if not IDENTIFIER_RE.fullmatch(value):
        raise ValueError(f"{name} must be a simple SQL identifier, got {value!r}")
    return value


def resolve_partition(raw: str) -> int | str:
    # Default to the current UTC year's partition. Set LOAD_PARTITION in config to
    # target another partition: a different year, a date string, a region code, and
    # so on. Digit-only values are treated as integers so they match a numeric
    # partition column (such as a Hive year) and still prune cleanly.
    if not raw:
        return datetime.now(timezone.utc).year
    return int(raw) if raw.lstrip("-").isdigit() else raw


if __name__ == "__main__":
    main()

requirements_txt:

duckdb==1.5.2

Private buckets need an account-level TYPE S3 secret (read by the engine, not env-injected); the default public bucket needs none.

Deploying a Template

1. Create any required TYPE flights secret on a read-write connection (query_rw, the UI, or a direct connection — the read-only query tool rejects CREATE SECRET). 2. Create the flight without a schedule. With MCP:

create_flight(
  name = "postgres-nightly-mirror",
  source_code = <flight source above>,
  requirements_txt = "duckdb==1.5.2\ntenacity==9.0.0\n",
  md_secret_names = ["pg"],
  config = { "TARGET_DATABASE": "postgres_ingest", "EXCLUDED_SCHEMAS": "audit,scratch" },
)

With SQL, the same call is FROM MD_CREATE_FLIGHT(name := ..., source_code := ..., requirements_txt := ..., flight_secret_names := ['pg'], config := MAP {...}). 3. Trigger one on-demand run (run_flight), poll list_flight_runs until terminal, and read get_flight_logs. Fix issues with edit_flight_source. 4. Attach the schedule only after a clean run: update_flight(id, schedule_cron = "0 6 * * *") — cron is UTC, and schedule changes do not create a new version.

Choosing a Template

MotherDuck Flights Reference

Condensed from the MotherDuck Flights docs (concepts, key tasks, MCP tool pages, SQL function pages) and the live get_flight_guide output. When MCP is available, get_flight_guide is the runtime source of truth; this file is the offline summary.

Contents

What a Flight Is

A Flight is a Python program MotherDuck schedules and runs server-side, on demand or on a recurring schedule, with direct access to your databases. Each run gets its own isolated runtime that executes main() to completion and exits — there is no managed worker pool, queue, or distributed state. The flight reaches data like any DuckDB client: it opens an md: connection that routes through a Duckling.

Use a flight when a job should run unattended on MotherDuck, retry on a schedule, and keep a run history: ingest from external sources, refresh aggregates, run dbt, export Parquet/CSV to object storage, post scheduled alerts, reverse ETL, AI enrichment. Do not use one for interactive exploration (use the SQL editor or motherduck-query) or for orchestration spanning many external systems with complex dependencies — a dedicated orchestrator like Airflow or Prefect is still the better tool there. Fan-out is your job: a flight can trigger other flights (MD_RUN_FLIGHT) or thread-pool within a run.

Anatomy of a Flight

Runtime Environment

Each run executes this sequence: allocate a Python runtime → inject MOTHERDUCK_TOKEN, config keys, and secret params as env vars → pip install the requirements → execute main() capturing stdout+stderr → record status and logs.

The container is constrained:

• 2 CPU cores — a large thread pool will not speed things up.
• 16GB RAM ceiling — exceeding it kills the run. Prefer disk-buffered loading over holding datasets in memory.
• ~150GB scratch disk at `/tmp/` — use it for staging files and on-disk DuckDB databases; clean up between batches.
• It is a Linux process: subprocess works, apt-get install of Debian packages works (git, ffmpeg, Playwright). dlt and similar tools that write under HOME should set os.environ.setdefault("HOME", "/tmp").
• DuckDB extensions can be installed inside the flight's local DuckDB process (INSTALL postgres, INSTALL bigquery FROM community) — the no-runtime-extension rule applies to MotherDuck's server-side engine, not to the flight container.
• Python version, run timeout, and concurrency quotas are not publicly documented; treat long runs and parallel runs conservatively.

Preview caveat: flights run on shared compute infrastructure across tenants. During Preview, do not process, store, or log ePHI, payment card data, or other regulated or sensitive personal data in flights.

Authentication

The runtime provides a MotherDuck access token as the MOTHERDUCK_TOKEN env var, so duckdb.connect("md:") works with no credentials in code. By default it is the MotherDuck Flights token for your user; pin a specific token (for example a service-account token) by passing its label as access_token_name (md_token_name on MCP). Pick a label whose scope covers exactly the databases the flight needs — and no more.

Config vs Secrets

Both arrive as env vars; the difference is encryption and naming.

• Config: non-sensitive knobs (region, table names, batch sizes). Key REGION arrives as env var REGION. Not encrypted — never put API keys, passwords, or tokens here.
• Secrets: create once, reuse across flights:

CREATE SECRET api_secret IN motherduck (
  TYPE flights,
  PARAMS MAP { 'API_KEY': 'sk-...', 'API_HOST': 'api.example.com' }
);

Reference it with flight_secret_names = ["api_secret"]. Each param is injected as `<secret_name>_<PARAM>` — here api_secret_API_KEY and api_secret_API_HOST, not bare API_KEY. DuckDB lowercases unquoted secret names; keep param keys UPPERCASE. On naming conflicts, the last secret in the list wins. This prefixing is the most common flight-authoring bug; a robust pattern is to check the bare name first (local runs) and then scan os.environ for any key ending in _<PARAM> (deployed runs).

• CREATE SECRET must run on a read-write connection (query_rw, the UI, or a direct connection); the read-only MCP query tool rejects it. S3 access for private buckets is separate: an account-level TYPE S3 secret read by the engine, not env-injected.

Scheduling

schedule_cron is a standard 5-field cron expression in UTC:

*/15 * * * *      every 15 minutes
0 * * * *         hourly at :00
0 6 * * *         daily at 06:00 UTC
0 6 * * 1         every Monday at 06:00 UTC

Step syntax requires a base: */N or M-N/S; a bare /N is invalid. Omit schedule_cron to create an on-demand-only flight; on update_flight, pass "" to clear the schedule, omit to leave it unchanged. A schedule can be active or disabled; disabling does not delete it. Schedule changes are metadata-only (no new version).

Versioning and Update Semantics

• Content fields — source_code, requirements_txt, config, flight_secret_names, access_token_name — are immutable per version. Any change to them creates a new 1-indexed FlightVersion.
• name and schedule_cron changes do not create a version.
• update_flight is a PATCH: omitted fields are unchanged, and when you touch any content field the others are carried forward — send only what changes. But config and flight_secret_names are full replacements when sent, never merges.
• A run locks to the version current when it started; a mid-run update only affects the next run.
• Inspect history with list_flight_versions or get_flight(id, version) — use the run record's flight_version to read the exact source a failing run executed.

Runs, Logs, and Cancellation

Run lifecycle: PENDING → RUNNING → terminal SUCCEEDED | FAILED | CANCELLED (SQL surface prefixes these with RUN_STATUS_). run_flight returns immediately with the new run (sequential per-flight run_number); poll list_flight_runs (newest first) for completion. exit_code is 0 on success, NULL while in progress. Multiple concurrent runs of one flight are allowed.

• run_flight(id, config?) — the optional config is a per-run override merged over the stored config (provided keys win, only keys already defined on the flight can be set); the flight itself is unchanged. Use it for backfill dates or one-off parameter changes. Non-secret values only.
• get_flight_logs(id, run_number, max_bytes?) — combined stdout/stderr plus the full run record (status, exit_code, timing) in one call; truncation keeps the tail (max_bytes minimum 1024). Logs are available while RUNNING and after any terminal status.
• cancel_flight_run(id, run_number) — returns canceled: true on a successful transition; calling it on a terminal or nonexistent run is a tool error.

MCP Tool Reference

Call get_flight_guide (no arguments) first — it returns the current authoring guide.

MCP vs SQL Naming

The same operations exist as server-side SQL functions (FROM MD_CREATE_FLIGHT(...), MD_UPDATE_FLIGHT, MD_RUN_FLIGHT, MD_FLIGHTS(), MD_GET_FLIGHT, MD_GET_FLIGHT_VERSION, MD_GET_FLIGHT_LOGS, MD_DELETE_FLIGHT, ...). They are not available on local-only DuckDB connections. Name differences:

Resolve a flight by name in SQL with SELECT flight_id FROM MD_FLIGHTS() WHERE flight_name = ?.

Loading Data from a Flight

Match the strategy to volume:

Avoid: executemany() (row-by-row under the hood), many small INSERT round-trips (~50–100ms each), and uncompressed temp tables that blow the 16GB ceiling (if you must, CHECKPOINT periodically). Prefer one bulk multi-row INSERT with bound parameters when inserting from Python lists.

Ingestion and Transformation Patterns

• dlt (recommended for API/source ingestion): declarative pipelines with schema evolution, incremental loading, and a native MotherDuck destination. Set loader_file_format="parquet" and write_disposition="merge" + primary_key for idempotent loads. See the dlt template in FLIGHT_EXAMPLES.md.
• Postgres: INSTALL postgres; LOAD postgres; ATTACH '' AS pg (TYPE postgres, READ_ONLY) with credentials passed through libpq env vars, then one streaming CREATE OR REPLACE TABLE ... AS SELECT * FROM pg."schema"."table" per table — atomic, idempotent, bounded memory. See the Postgres template in FLIGHT_EXAMPLES.md.
• Other warehouses via community extensions: open a local DuckDB (duckdb.connect(":memory:", config={"allow_community_extensions": True})), INSTALL bigquery FROM community; LOAD bigquery; LOAD motherduck; ATTACH 'md:' — extensions unsupported server-side still work inside the flight container. Keep heavy compute in the source warehouse or MotherDuck; the flight just moves data.
• Transformation/refresh: recompute heavy aggregates on a cron and write to a target table dashboards read; for DAG-shaped transformation, run dbt with the dbt-duckdb adapter inside the flight.
• Audit ledger: append one row per run (run_at TIMESTAMPTZ first) to a small tracker table so every run leaves a queryable trail.

Production Checklist

1. One successful on-demand run with logs reviewed before any schedule is attached. 2. Service-account token via access_token_name, scoped to only the target databases. 3. All sensitive values in TYPE flights secrets; nothing sensitive in config or source. 4. requirements_txt fully pinned (supply-chain note from the docs: flights do not scan your code or dependencies — avoid untrusted packages). 5. Idempotent writes and IF NOT EXISTS bootstrap so reruns and first runs both succeed. 6. Non-zero exit on failure (raise, or sys.exit(1)) so the run reports FAILED instead of silently succeeding.

Troubleshooting