Writing robust data pipelines with AILivestream July 22

Skip to main content

Connect from Python via Postgres endpoint

Preview
This feature is in preview and is subject to change.

You can query MotherDuck from Python using standard PostgreSQL client libraries. No DuckDB installation is required. This guide covers psycopg2 and psycopg (v3).

For connection parameters, SSL options, and limitations, see the Postgres Endpoint reference.

Prerequisites​

You need a MotherDuck access token. Set it as an environment variable:

export MOTHERDUCK_TOKEN="your_token_here"

Connect​

# /// script
# dependencies = ["psycopg"]
# ///

import os
import psycopg

with psycopg.connect(
    host="pg.us-east-1-aws.motherduck.com",  # or us-west-2-aws, eu-central-1-aws, eu-west-1-aws, ap-northeast-1-aws, or ap-southeast-2-aws
    port=5432,
    dbname="md:",
    user="postgres",
    password=os.environ["MOTHERDUCK_TOKEN"],
    sslmode="verify-full",
    sslrootcert="system",  # available in libpq 16+
) as conn:
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT title, score
            FROM sample_data.hn.hacker_news
            WHERE type = 'story'
            ORDER BY score DESC
            LIMIT 5
            """
        )
        for row in cur:
            print(row)

You can also use a connection URI:

import os
import psycopg

token = os.environ["MOTHERDUCK_TOKEN"]
with psycopg.connect(
    f"postgresql://postgres:{token}@pg.us-east-1-aws.motherduck.com:5432/md:?sslmode=verify-full&sslrootcert=system"
) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT current_database()")
        print(cur.fetchone())

Use md: as the database name, or replace it with a specific database name such as sample_data.

Connection pooling and timeouts​

Use a connection pool in production. With psycopg v3, install pool support:

pip install "psycopg[pool]"

Then create one pool per application process:

import os
from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    conninfo=(
        "host=pg.us-east-1-aws.motherduck.com "
        "port=5432 "
        "dbname=md: "
        "user=postgres "
        "sslmode=verify-full "
        "sslrootcert=system"
    ),
    kwargs={"password": os.environ["MOTHERDUCK_TOKEN"]},
    min_size=0,
    max_size=10,
    timeout=5,
    max_idle=30,
    max_lifetime=300,
)

with pool.connection() as conn:
    with conn.cursor() as cur:
        cur.execute(
            "SELECT title, score FROM sample_data.hn.hacker_news WHERE type='story' LIMIT 10"
        )
        print(cur.fetchall())

timeout=5 fails fast when the pool cannot provide a connection. max_idle=30 closes unused connections quickly when the pool can shrink, and max_lifetime=300 periodically replaces long-lived connections. The pool context manager returns healthy connections to the pool and discards broken ones. If you catch database errors inside the block, roll back failed transactions before reusing the connection.

statement_timeout is not supported through the Postgres endpoint today. For sync psycopg code, use a client-side timer that calls conn.cancel():

import threading
import psycopg

with pool.connection() as conn:
    with conn.cursor() as cur:
        timer = threading.Timer(60, conn.cancel)
        timer.start()
        try:
            cur.execute("SELECT count(*) FROM sample_data.hn.hacker_news")
            print(cur.fetchone())
        except psycopg.errors.QueryCanceled as exc:
            conn.rollback()
            raise TimeoutError("MotherDuck query exceeded 60 seconds") from exc
        finally:
            timer.cancel()

For async psycopg code, wrap the query in asyncio.timeout(...); psycopg sends cancellation when the task is cancelled.

Loading data from Python​

For loading through the Postgres endpoint, the recommended pattern is server-side reads from remote storage:

  • Use psycopg or SQLAlchemy to execute CREATE TABLE AS SELECT or INSERT INTO ... SELECT.
  • Point read_parquet, read_csv, or read_json at S3, GCS, R2, Azure, or HTTPS.
  • Set MD_RUN = REMOTE on those file reads.

Example with SQLAlchemy:

import os
from sqlalchemy import create_engine, text

engine = create_engine(
    "postgresql+psycopg://postgres:@pg.us-east-1-aws.motherduck.com:5432/md:",
    connect_args={
        "password": os.environ["MOTHERDUCK_TOKEN"],
        "sslmode": "require",
    },
)

with engine.begin() as conn:
    conn.execute(
        text(
            """
        CREATE OR REPLACE TABLE my_db.main.weather_events AS
        SELECT *
        FROM read_csv(
            'https://raw.githubusercontent.com/duckdb/duckdb-web/main/data/weather.csv',
            HEADER = true,
            AUTO_DETECT = true,
            MD_RUN = REMOTE
        )
        """
        )
    )

The following patterns are not supported from Python over the Postgres endpoint:

  • COPY ... FROM '/local/file.csv'
  • cursor.copy(...) / COPY FROM STDIN
  • psql \copy
  • MD_RUN = LOCAL
  • SQLAlchemy's default executemany path for bulk ingest

If the rows exist only in application memory and the volume is modest, prefer explicit multi-values INSERT statements. For large local bulk loads, switch to a DuckDB client path instead.

See Loading data through the Postgres endpoint for the full decision guide.

SSL notes​

  • psycopg (v3) wraps libpq and supports sslrootcert=system directly.
  • psycopg2 bundles its own statically linked OpenSSL, so sslrootcert=system is not supported. Use the certifi package to point to CA certificates, or download the ISRG Root X1 certificate and set sslrootcert to its path.

For more details on SSL options, see SSL and certificate verification.