Got it — here’s a minimal, working set of files for Prefect 3.x that: - Bakes your flow code into a Docker image - Uses a custom Dockerfile that copies

pyproject.toml

and installs with

uv

- Is a good replacement for the defaults you get from

prefect init

Notes: - Assumes a Docker work pool named

docker-pool

and a flow at

flows/hello.py:hello_flow

- Uses the

prefect-docker

deployment build step to build and (optionally) push your image - Includes a

pull

step to set the working directory inside the container prefect.yaml

prefect-version: ">=3.0.0"
name: baked-image-project

# Build a Docker image that contains your code and dependencies
build:
  - prefect.deployments.steps.run_shell_script:
      id: git-sha
      script: git rev-parse --short HEAD || echo "local"
      stream_output: false

  - prefect_docker.deployments.steps.build_docker_image:
      requires: prefect-docker>=0.3.1
      id: build-image
      image_name: your-registry/your-project
      tag: "{{ git-sha.stdout }}"
      dockerfile: Dockerfile
      context: .
      pull: true

# Optional: push the image to a registry (remove if not needed)
push:
  - prefect_docker.deployments.steps.push_docker_image:
      requires: prefect-docker>=0.3.1
      image_name: "{{ build-image.image_name }}"
      tag: "{{ build-image.tag }}"
      # If using a Docker Registry Credentials block, uncomment the next line and set your block name:
      # credentials: "{{ prefect.blocks.docker-registry-credentials.my-registry-creds }}"

# For baked images, just set the working directory at runtime
pull:
  - prefect.deployments.steps.set_working_directory:
      directory: /opt/prefect

deployments:
  - name: hello-baked
    entrypoint: flows/hello.py:hello_flow
    description: "Baked-into-image deployment using uv + pyproject.toml"
    work_pool:
      name: docker-pool
      job_variables:
        # Use the image we just built/pushed
        image: "{{ build-image.image }}"
        # Example: set env vars for the runtime container
        env:
          PYTHONUNBUFFERED: "1"

Dockerfile

# Use Prefect's base image so Prefect is already installed
FROM prefecthq/prefect:3-latest

# Recommended workdir for Prefect projects
WORKDIR /opt/prefect

# Install uv and any OS deps you need
# (uv is a fast pip-compatible installer; prefect is already present in this image)
RUN pip install --no-cache-dir -U uv

# Copy dependency metadata first (improves layer caching if only deps change)
COPY pyproject.toml ./
# If you keep a uv.lock file, copy it as well for reproducible installs
# COPY uv.lock ./

# Copy your source code last
COPY . .

# Install your project with uv from pyproject.toml (editable install is typical for apps)
# Ensure your pyproject has your runtime deps (including prefect) declared
RUN uv pip install --system -e .

# The worker will set the command for running flow runs, but this entrypoint is safe
ENTRYPOINT ["python", "-m", "prefect.engine"]

docker-compose.yml (worker only)

version: "3.9"

services:
  prefect-worker:
    image: prefecthq/prefect:3-latest
    container_name: prefect-docker-worker
    environment:
      # Point to Prefect Cloud or a local server
      # For Prefect Cloud (recommended):
      # PREFECT_API_URL: <https://api.prefect.cloud/api/accounts/<account_id>/workspaces/<workspace_id>>
      # PREFECT_API_KEY: <your_api_key>

      # For local server (if you're running it):
      # PREFECT_API_URL: <http://host.docker.internal:4200/api>
    command: >
      prefect worker start
      --pool docker-pool
    volumes:
      # Allow worker to start Docker job containers on the same host
      - /var/run/docker.sock:/var/run/docker.sock
    restart: unless-stopped

How to use this (quick start) - Create a Docker work pool (run once):

prefect work-pool create "docker-pool" --type docker

- Build and register the deployment from your project root: ```

prefect deploy --prefect-file prefect.yaml --name hello-baked --no-prompt

- Start the worker with Docker Compose:

docker compose up -d

- Kick off a run:
  - In Prefect UI, run the deployment “hello-baked”
  - Or CLI:

prefect deployment run 'hello_flow/hello-baked' ``` Tips when changing from prefect init - Replace the default scaffold’s build/pull sections with the ones above - Keep your flow entrypoint in

entrypoint: flows/hello.py:hello_flow

- Ensure your

pyproject.toml

includes Prefect 3 and your runtime dependencies - If you don’t push to a registry, the image must exist locally wherever your Docker worker runs If you want me to tailor these to your actual repo names, flow path, and registry (e.g., GHCR/ECR), share those details and I’ll adjust the files accordingly.

Short answer: - With docker-git, your repo is cloned at runtime under the container’s current working directory (default is

/opt/prefect

), into a folder named after the repo (and sometimes suffixed with the branch). - If you add a

set_working_directory

pull step to that clone directory, then writing to

data/

in your flow will go to

/opt/prefect/<repo-name>/data

(or

/opt/prefect/<repo-name>-<branch>/data

). - If you don’t set the working directory after cloning,

data/

will resolve to whatever the container’s CWD is (likely

/opt/prefect/data

), but imports may fail without the CWD set to the clone. Why you didn’t see anything: - The file was almost certainly written inside the flow-run container at

/opt/prefect/<repo-name>/data

(ephemeral), not on your host. - Mounting the volume on the worker container does not persist files from the flow-run container. You must mount the volume on the job container via

job_variables.volumes

in the deployment or work pool. - Your example mount path looks off:

...:/opt/prefect/prefect_demo/opt/prefect/data

has a duplicated segment. What to do 1) Explicitly set the working directory after cloning, so relative writes go where you expect Add this to your

prefect.yaml

pull steps:

pull:
  - prefect.deployments.steps.git_clone:
      id: clone
      repository: <https://github.com/your-org/prefect_demo.git>
      branch: main
  - prefect.deployments.steps.set_working_directory:
      directory: "{{ clone.directory }}"

Now

data/

resolves to

/opt/prefect/prefect_demo/data

(or

/opt/prefect/prefect_demo-main/data

if Prefect added the branch suffix). 2) Mount a volume on the flow-run container (not just the worker) Add a volume to the deployment’s job variables so the container that executes your flow writes to your host:

deployments:
  - name: my-deployment
    entrypoint: flows/flow.py:my_flow
    work_pool:
      name: docker-pool
      job_variables:
        volumes:
          # If you used set_working_directory as above:
          - "/home/choupoheng/repos/prefect_demo/data:/opt/prefect/prefect_demo/data"
          # If Prefect includes the branch in the folder name, either:
          # - mount the entire /opt/prefect and keep your code writing to /opt/prefect/<repo>/data
          # - or switch your code to a fixed absolute path (see option 3 below)

3) Consider a stable, absolute data path (avoids repo/branch suffix issues) Instead of relying on the repo folder name, write to a fixed path and mount that: - In your code: write to

/mnt/data/...

(or pull

DATA_DIR

from an env var). - In deployment job variables:

work_pool:
  name: docker-pool
  job_variables:
    env:
      DATA_DIR: "/mnt/data"
    volumes:
      - "/home/choupoheng/repos/prefect_demo/data:/mnt/data"

4) Quick runtime check Add this to your flow to see where files are going:

import os
from pathlib import Path
print("CWD:", os.getcwd())
print("data abs:", Path("data").resolve())

5) Minor Dockerfile note about uv

uv sync

creates a venv by default; your container may not activate it. If you want system-wide installs in the Prefect base image, prefer:

FROM prefecthq/prefect:3.6.4-python3.11
WORKDIR /opt/prefect
COPY pyproject.toml uv.lock ./
RUN pip install -U uv && uv pip install --system -e .

Common pitfalls to avoid - Mounting volumes only on the worker container. The flow runs in a separate job container started by the worker; mount volumes via

work_pool.job_variables.volumes

or your work pool’s base job template. - Mount path typos. For your case, correct would be something like: -

/home/choupoheng/repos/prefect_demo/data:/opt/prefect/prefect_demo/data

not -

/home/choupoheng/repos/prefect_demo/data:/opt/prefect/prefect_demo/opt/prefect/data

If you share your repo URL/name and branch, I can give you the exact in-container path and the precise volumes line to use. Helpful docs: - Deployments

- Work pools - Docker work pool

For Harbor, set the image_name to your registry host + project + repo (no scheme), then use tag as usual. Format: - image_name: harbor.example.com/&lt;project&gt;/&lt;repository> - tag: <tag> Examples:

build:
  - prefect_docker.deployments.steps.build_docker_image:
      id: build_image
      requires: prefect-docker>=0.3.1
      image_name: <http://harbor.mycompany.com/data-platform/prefect-demo|harbor.mycompany.com/data-platform/prefect-demo>
      tag: test-v1
      dockerfile: Dockerfile

If your Harbor runs on a non-standard port:

image_name: <http://harbor.mycompany.com:5000/data-platform/prefect-demo|harbor.mycompany.com:5000/data-platform/prefect-demo>

Optional push step with Harbor credentials (recommended):

push:
  - prefect_docker.deployments.steps.push_docker_image:
      requires: prefect-docker>=0.3.1
      image_name: "{{ build_image.image_name }}"
      tag: "{{ build_image.tag }}"
      credentials: "{{ prefect.blocks.docker-registry-credentials.harbor_prod }}"

Notes: - Use lowercase for repository names; tags can include letters, numbers, dots, underscores, and hyphens (e.g., test-v1, 1.2.3). - Do not include https:// in image_name (only host/project/repo). - For Harbor robot accounts, create a Docker Registry Credentials block with: - registry_url: https://harbor.mycompany.com - username: robot$project+robotname - password: <robot token> - If Harbor uses a self-signed cert or is “insecure,” configure your Docker daemon accordingly so build/push can succeed. Docs: - Prefect Docker integration - Deployments

You’ve got two places to wire Harbor credentials so Docker workers can pull private images: A) Work pool (recommended default for all deployments on that pool) - Create a Docker Registry Credentials block for Harbor:

from prefect_docker import DockerRegistryCredentials

  creds = DockerRegistryCredentials(
      username="robot$project+bot",
      password="...robot_token...",
      registry_url="<http://harbor.mycompany.com|harbor.mycompany.com>",  # no scheme
      reauth=True
  )
  await creds.save("harbor-prod", overwrite=True)

- Add the block to your Docker work pool’s base job template (UI or CLI). In the pool template JSON, set:

"registry_credentials": "{{ prefect.blocks.docker-registry-credentials.harbor-prod }}"

This makes every job started by that pool log in to Harbor before pulling the image. B) Per-deployment override (only for one deployment) - In prefect.yaml:

deployments:
    - name: my-harbor-deploy
      entrypoint: flows/main.py:my_flow
      work_pool:
        name: docker-pool
        job_variables:
          image: <http://harbor.mycompany.com/myproject/myimage:test-v1|harbor.mycompany.com/myproject/myimage:test-v1>
          registry_credentials: "{{ prefect.blocks.docker-registry-credentials.harbor-prod }}"
          # optional pull policy
          image_pull_policy: Always  # or IfNotPresent / Never

Build/push steps (if you’re building the image in your pipeline) - Build to Harbor:

build:
    - prefect_docker.deployments.steps.build_docker_image:
        id: build_image
        requires: prefect-docker>=0.3.1
        image_name: <http://harbor.mycompany.com/myproject/myimage|harbor.mycompany.com/myproject/myimage>
        tag: test-v1
        dockerfile: Dockerfile

- Push with the same credentials block:

push:
    - prefect_docker.deployments.steps.push_docker_image:
        requires: prefect-docker>=0.3.1
        image_name: "{{ build_image.image_name }}"
        tag: "{{ build_image.tag }}"
        credentials: "{{ prefect.blocks.docker-registry-credentials.harbor-prod }}"

Notes and tips - image_name format for Harbor:

<http://harbor.mycompany.com/<project>/<repo|harbor.mycompany.com/<project>/<repo>>

(no https://; include port if non-standard). - The worker does not need

docker login

on the host; it logs in programmatically using the credentials block before pulling. - If Harbor uses a self-signed cert, configure the Docker daemon on the worker host with the CA or mark the registry as insecure (daemon.json), then restart Docker. - Use

image_pull_policy

wisely: - IfNotPresent for pinned tags (immutable) - Always for mutable tags like latest - You can set

registry_credentials

once in the work pool’s base job template so deployments don’t need to repeat it. If you share your pool name and the exact Harbor image (host/project/repo:tag), I can give you a drop-in prefect.yaml snippet for your setup.

Got it — build/push to Harbor works with your Docker Registry Credentials block, but the flow run fails because the Docker worker can’t pull the private image at runtime. That usually means the worker job wasn’t given the registry credentials (push step had them; runtime doesn’t). Do this: 1) Attach the credentials to the job that runs your flow - Either add them to your work pool’s base job template (applies to all deployments in the pool), or set them per deployment via job_variables. Per-deployment (prefect.yaml):

deployments:
  - name: my-harbor-deploy
    entrypoint: flows/main.py:my_flow
    work_pool:
      name: docker-pool
      job_variables:
        image: <http://harbor.mycompany.com/myproject/myimage:test-v1|harbor.mycompany.com/myproject/myimage:test-v1>
        registry_credentials: "{{ prefect.blocks.docker-registry-credentials.harbor-prod }}"
        image_pull_policy: Always  # or IfNotPresent for pinned tags

If your block name contains a dot (e.g., docker.creds), use bracket syntax:

registry_credentials: "{{ prefect.blocks.docker-registry-credentials['docker.creds'] }}"

Pool-wide default (recommended so you don’t repeat config): - In the Prefect UI, open your Docker work pool → Edit Base Job Template → add:

"registry_credentials": "{{ prefect.blocks.docker-registry-credentials.harbor-prod }}"

Then your deployments don’t need to set it. 2) Verify your pool config and image name - Check the pool template:

prefect work-pool inspect docker-pool

Ensure you see registry_credentials resolved and that the deployment sets a fully qualified image:

<http://harbor.mycompany.com/<project>/<repo>:<tag|harbor.mycompany.com/<project>/<repo>:<tag>>

Include port if needed:

<http://harbor.mycompany.com:5000/|harbor.mycompany.com:5000/>...

3) If Harbor uses a custom/self-signed cert - The worker host’s Docker daemon must trust it: - Place CA at:

/etc/docker/certs.d/harbor.mycompany.com/ca.crt

(and include :port directory if using a port) - Or configure `/etc/docker/daemon.json`:

{
  "insecure-registries": [
    "<http://harbor.mycompany.com:5000|harbor.mycompany.com:5000>"
  ]
}

- Then restart Docker. 4) Start the worker with the right pool/type

prefect worker start --type docker --pool docker-pool

5) Check the exact error to confirm - In the flow run logs (or worker logs), common messages: - “authentication required” → credentials not applied or registry_url mismatch - “x509: certificate signed by unknown authority” → trust/cert issue on worker host - “manifest unknown” → image/tag path wrong If you can share the error text, I can pinpoint the fix. Reference docs: - Docker work pool - Prefect Docker integration

Great — since you already created a Postgres block in the Prefect UI, the easiest way to connect and save data is with the prefect-sqlalchemy integration. Below are concise, production-ready examples for both sync and async flows that load your UI block by name and perform create/insert/select. Install (in your image or env)

pip install "prefect-sqlalchemy>=0.6.0" "psycopg2-binary>=2.9" "asyncpg>=0.29"

Sync example (psycopg2)

from prefect import flow, task
from prefect_sqlalchemy import SqlAlchemyConnector

@task
def write_and_read_users():
    # Load the block you created in the UI (use your exact block name)
    db = SqlAlchemyConnector.load("postgres-prod")

    # Use as a context manager to manage engine/connection lifecycle
    with db:
        # Create a table if it doesn't exist
        db.execute("""
            CREATE TABLE IF NOT EXISTS users (
              id SERIAL PRIMARY KEY,
              name VARCHAR(100),
              email VARCHAR(255) UNIQUE
            );
        """)

        # Upsert-like pattern (simple example)
        db.execute(
            "INSERT INTO users (name, email) VALUES (:name, :email) "
            "ON CONFLICT (email) DO NOTHING;",
            parameters={"name": "Alice", "email": "<mailto:alice@example.com|alice@example.com>"}
        )

        # Bulk insert
        db.execute_many(
            "INSERT INTO users (name, email) VALUES (:name, :email) "
            "ON CONFLICT (email) DO NOTHING;",
            seq_of_parameters=[
                {"name": "Bob", "email": "<mailto:bob@example.com|bob@example.com>"},
                {"name": "Charlie", "email": "<mailto:charlie@example.com|charlie@example.com>"},
            ],
        )

        # Query
        rows = db.fetch_all("SELECT id, name, email FROM users ORDER BY id;")
        return rows

@flow
def sync_postgres_flow():
    rows = write_and_read_users()
    for r in rows:
        print(r)

if __name__ == "__main__":
    sync_postgres_flow()

Async example (asyncpg)

import asyncio
from prefect import flow, task
from prefect_sqlalchemy import SqlAlchemyConnector

@task
async def write_and_read_events():
    db = SqlAlchemyConnector.load("postgres-async")  # your async-configured block

    async with db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS events (
              id SERIAL PRIMARY KEY,
              event_name VARCHAR(100),
              ts TIMESTAMP DEFAULT NOW()
            );
        """)

        await db.execute(
            "INSERT INTO events (event_name) VALUES (:name);",
            parameters={"name": "user_login"},
        )

        await db.execute_many(
            "INSERT INTO events (event_name) VALUES (:name);",
            seq_of_parameters=[{"name": "user_logout"}, {"name": "data_export"}],
        )

        rows = await db.fetch_all("SELECT id, event_name, ts FROM events ORDER BY id;")
        return rows

@flow
async def async_postgres_flow():
    rows = await write_and_read_events()
    for r in rows:
        print(r)

if __name__ == "__main__":
    asyncio.run(async_postgres_flow())

Notes and tips - Block name: replace

postgres-prod

/

postgres-async

with the exact name of your UI block. - The same SqlAlchemyConnector block can be configured with either: - Connection string:

<postgresql+psycopg2://user:pass@host:5432/db>

- Connection components (host, port, db, username, password), sync or async driver - Use

with db:

for sync blocks and

async with db:

for async blocks. Mixing them will raise runtime errors. - For large reads, set

db.fetch_size

and use

fetch_many

in a loop. - In Docker/K8s, ensure networking/resolution to your Postgres host, and open port 5432. Docs: - prefect-sqlalchemy integration - Blocks in Prefect

Great—here are two solid patterns to load a Parquet dataset into Postgres using your Prefect UI block. Install (if not already in your image/env)

pip install "prefect-sqlalchemy>=0.6.0" "pandas>=2.0" "pyarrow>=14" "psycopg2-binary>=2.9"

Approach A (simple, good for small/medium files): pandas -> to_sql

from prefect import flow, task
from prefect_sqlalchemy import SqlAlchemyConnector
from sqlalchemy import inspect
import pandas as pd

@task
def read_parquet(path: str) -> pd.DataFrame:
    # Reads local path or remote URLs supported by fsspec (if configured)
    return pd.read_parquet(path)

@task
def load_df_to_postgres(df: pd.DataFrame, table: str, schema: str, block_name: str):
    db = SqlAlchemyConnector.load(block_name)
    with db:
        engine = db.get_engine()

        # Create table if it doesn’t exist (schema-only)
        insp = inspect(engine)
        if not insp.has_table(table, schema=schema):
            df.head(0).to_sql(
                name=table,
                con=engine,
                schema=schema,
                if_exists="append",
                index=False,
            )

        # Append data in chunks
        df.to_sql(
            name=table,
            con=engine,
            schema=schema,
            if_exists="append",
            index=False,
            chunksize=10_000,
            method="multi",  # batch insert
        )

@flow
def load_parquet_to_postgres(
    parquet_path: str,
    table: str,
    schema: str = "public",
    block_name: str = "postgres-prod",  # replace with your UI block name
):
    df = read_parquet(parquet_path)
    load_df_to_postgres(df, table, schema, block_name)

if __name__ == "__main__":
    load_parquet_to_postgres(
        parquet_path="data/my_dataset.parquet",
        table="my_table",
        schema="public",
        block_name="postgres-prod"
    )

Approach B (faster for large files): COPY via psycopg2 - This streams CSV to Postgres with COPY. Table must exist (we’ll create it if missing).

from prefect import flow, task
from prefect_sqlalchemy import SqlAlchemyConnector
from sqlalchemy import inspect
import pandas as pd
from io import StringIO

@task
def read_parquet(path: str) -> pd.DataFrame:
    return pd.read_parquet(path)

@task
def create_table_if_missing(df: pd.DataFrame, table: str, schema: str, block_name: str):
    db = SqlAlchemyConnector.load(block_name)
    with db:
        engine = db.get_engine()
        insp = inspect(engine)
        if not insp.has_table(table, schema=schema):
            # Create schema from df columns (no rows)
            df.head(0).to_sql(
                name=table,
                con=engine,
                schema=schema,
                if_exists="append",
                index=False,
            )

@task
def copy_via_psycopg2(df: pd.DataFrame, table: str, schema: str, block_name: str):
    # Convert to CSV (no header) for COPY
    buf = StringIO()
    df.to_csv(buf, index=False, header=False)
    buf.seek(0)

    db = SqlAlchemyConnector.load(block_name)
    with db:
        engine = db.get_engine()
        # Use raw DBAPI connection to access psycopg2 copy_expert
        with engine.raw_connection() as raw:
            with raw.cursor() as cur:
                cur.copy_expert(
                    sql=f"COPY {schema}.{table} FROM STDIN WITH (FORMAT CSV)",
                    file=buf,
                )
            raw.commit()

@flow
def load_parquet_fast(
    parquet_path: str,
    table: str,
    schema: str = "public",
    block_name: str = "postgres-prod",
):
    df = read_parquet(parquet_path)
    create_table_if_missing(df, table, schema, block_name)
    copy_via_psycopg2(df, table, schema, block_name)

if __name__ == "__main__":
    load_parquet_fast(
        parquet_path="data/my_dataset.parquet",
        table="my_table",
        schema="public",
        block_name="postgres-prod"
    )

Notes

- Replace

postgres-prod

with the exact name of your SQLAlchemy Connector block in the Prefect UI. - Approach A auto-creates the table (basic schema inference via pandas/SQLAlchemy). For stricter control, define the table with explicit SQLAlchemy types first, then use

if_exists="append"

. - Approach B is much faster for large datasets, but the table must exist; the snippet auto-creates it if missing using df.head(0).to_sql, then uses COPY for the data. - If your dataset is huge and memory is tight, consider reading Parquet in row groups/chunks with pyarrow and writing in chunks (call the write task repeatedly). - In Docker/K8s, ensure your flow container can reach the Postgres host and port 5432, and your block’s connection string/driver matches your environment. Docs: - prefect-sqlalchemy integration - pandas to_sql - SQLAlchemy

Awesome — here’s a clear, step‑by‑step path to use prefect-sqlalchemy in Prefect 3.x to connect to Postgres and run queries. Step 0) Prereqs - Prefect 3.x - A running Postgres you can reach from your flow’s runtime (Docker/K8s networking!) - Block you’ll create: “SQLAlchemy Connector” Step 1) Install packages - In your dev env or Docker image:

pip install "prefect-sqlalchemy>=0.6.0" "psycopg2-binary>=2.9"  # sync driver
# Optional async support:
pip install "asyncpg>=0.29"

- If you build images with uv/pyproject.toml, add these to your dependencies and build. Step 2) Create the block in Prefect UI - Go to Blocks → + Create Block → SQLAlchemy Connector - Choose “Connection components” - Driver: “PostgreSQL psycopg2” (sync) or “PostgreSQL asyncpg” (async) - Host: your DB host - Port: 5432 - Database: your_db - Username / Password - Save it with a name (e.g., postgres-prod) - Tip: You can also use a full URL instead, e.g.

<postgresql+psycopg2://user:pass@host:5432/db>

Step 3) Sanity check the connection (tiny script)

from prefect_sqlalchemy import SqlAlchemyConnector

db = SqlAlchemyConnector.load("postgres-prod")  # use your block name
with db:
    one = db.fetch_one("SELECT 1;")
    print("SELECT 1 ->", one)

Step 4) Use it in a Prefect flow (sync example)

from prefect import flow, task
from prefect_sqlalchemy import SqlAlchemyConnector

@task
def write_and_read():
    db = SqlAlchemyConnector.load("postgres-prod")
    with db:
        db.execute("""
            CREATE TABLE IF NOT EXISTS users (
              id SERIAL PRIMARY KEY,
              name VARCHAR(100),
              email VARCHAR(255) UNIQUE
            );
        """)
        db.execute(
            "INSERT INTO users (name, email) VALUES (:name, :email) "
            "ON CONFLICT (email) DO NOTHING;",
            parameters={"name": "Alice", "email": "<mailto:alice@example.com|alice@example.com>"}
        )
        rows = db.fetch_all("SELECT id, name, email FROM users ORDER BY id;")
        return rows

@flow
def postgres_flow():
    rows = write_and_read()
    for r in rows:
        print(r)

if __name__ == "__main__":
    postgres_flow()

Step 5) Async variant (if your block uses asyncpg)

import asyncio
from prefect import flow, task
from prefect_sqlalchemy import SqlAlchemyConnector

@task
async def write_and_read_async():
    db = SqlAlchemyConnector.load("postgres-async")  # your async block
    async with db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS events (
              id SERIAL PRIMARY KEY,
              event_name VARCHAR(100),
              ts TIMESTAMP DEFAULT NOW()
            );
        """)
        await db.execute(
            "INSERT INTO events (event_name) VALUES (:name);",
            parameters={"name": "user_login"}
        )
        rows = await db.fetch_all("SELECT id, event_name, ts FROM events ORDER BY id;")
        return rows

@flow
async def postgres_flow_async():
    rows = await write_and_read_async()
    for r in rows:
        print(r)

if __name__ == "__main__":
    asyncio.run(postgres_flow_async())

Step 6) Deployment and Docker tips - Ensure your image includes the drivers: - psycopg2-binary for sync, asyncpg for async - If your DB is not public, confirm network reachability from the job container (Docker network, K8s Service, firewall). - If you need SSL/connect_args, add them in the block’s “Connect args” field in the UI (e.g.,

{"sslmode": "require"}

). Step 7) Loading Parquet into Postgres (quick example with pandas) ``` from prefect import flow, task from prefect_sqlalchemy import SqlAlchemyConnector import pandas as pd @task def read_parquet(path: str) -> pd.DataFrame: return pd.read_parquet(path) @task def load_df(df: pd.DataFrame, table: str, schema: str, block_name: str): db = SqlAlchemyConnector.load(block_name) with db: engine = db.get_engine() df.to_sql( name=table, con=engine, schema=schema,

if_exists="append", index=False, chunksize=10_000, method="multi", ) @flow def parquet_to_postgres(parquet_path: str, table: str, schema: str = "public", block_name: str = "postgres-prod"): df = read_parquet(parquet_path) load_df(df, table, schema, block_name) ``` Troubleshooting checklist - ImportError: install the correct driver (

psycopg2-binary

for sync,

asyncpg

for async). - RuntimeError “cannot be run asynchronously/synchronously”: match context manager to driver (

with

for sync,

async with

for async). - Connection refused/timeouts: verify host/port and container networking. - SSL required: set connect_args in the block (e.g.,

{"sslmode": "require"}

). Docs - prefect-sqlalchemy integration - Blocks in Prefect If you share your block name and whether you prefer sync or async, I can tailor the exact snippet for your setup.

Here’s how pandas.to_sql uses chunksize=10000 and method="multi" under the hood, especially relevant when loading into Postgres via prefect-sqlalchemy. What chunksize=10000 does - Pandas splits your DataFrame into batches of up to 10,000 rows each. - It then inserts one batch at a time. - If your DataFrame has N rows, pandas will perform ceil(N / 10,000) insert operations. What method="multi" does - Within each batch, pandas builds a single INSERT statement with many VALUES tuples and sends it in one DB call (when the backend supports it). - For example, with 3 rows and 2 columns:

INSERT INTO schema.table (col1, col2)
  VALUES (%s, %s), (%s, %s), (%s, %s)

- This reduces round trips and is usually much faster than issuing one INSERT per row. How it plays with Postgres (psycopg2) - You’ll typically get one big multi-VALUES INSERT per chunk. - This is fast, but you must respect Postgres’s parameter limit (~65,535 parameters per statement). - Each row contributes one placeholder per column. - Rule of thumb: chunksize <= floor(65535 / number_of_columns) - Example: 30 columns → 65535 / 30 ≈ 2184 max rows per statement. Using chunksize=10,000 here would exceed the parameter limit and fail. Reduce chunksize accordingly. Transactions and commits - to_sql runs inside a transaction (via SQLAlchemy). If a batch fails, it can roll back that transaction. - Very large chunks mean larger single statements; if you hit timeouts or parameter limits, lower chunksize. Performance tuning tips - Start with chunksize between 1,000 and 10,000, then tune: - If you hit parameter limit errors or timeouts, decrease chunksize. - If you see many small fast calls and want fewer round trips, increase chunksize (within limits). - For very large loads, Postgres COPY is typically faster than INSERT: - Use psycopg2’s COPY (e.g., via

copy_expert

) or a callable to_sql method. - Alternative to method="multi": - method=None (default) may use executemany depending on the backend, but usually slower than multi. - method="single" forces one-row-per-statement; generally slow for big loads. - method=callable: you can plug in psycopg2.extras.execute_values for even better performance. Optional: using a callable to leverage execute_values

from psycopg2.extras import execute_values

def pg_execute_values(table, conn, keys, data_iter):
    cols = ",".join(keys)
    schema = f"{table.schema}." if table.schema else ""
    sql = f"INSERT INTO {schema}{table.name} ({cols}) VALUES %s"
    # conn here is a SQLAlchemy Connection; conn.connection is the raw DBAPI connection
    with conn.connection.cursor() as cur:
        execute_values(cur, sql, (tuple(row[k] for k in keys) for row in data_iter), page_size=10000)

Then call:

df.to_sql(
    name="my_table",
    con=engine,
    schema="public",
    if_exists="append",
    index=False,
    method=pg_execute_values,
    chunksize=10000,
)

Bottom line - chunksize controls how many rows per batch. - method="multi" packs each batch into a single multi-VALUES INSERT, which is fast. - Watch Postgres’s parameter limit: rows_per_batch × num_columns <= 65535. - For very large/fast loads, consider COPY-based loading instead.

Shortest answer: - Exactly 1 field is strictly required by the block schema:

driver

. - For a real Postgres connection, you’ll typically need to enter 5 fields in the UI:

driver

,

host

,

database

,

username

,

password

. -

port

is only needed if it’s not the default (5432).

query

is optional (e.g.,

sslmode=require

). Details (Connection Components mode in the UI) - Required to select: - driver: choose one - Sync: “PostgreSQL psycopg2” - Async: “PostgreSQL asyncpg” - Typically needed to connect: - host: e.g.,

db.mycompany.local

(or a Docker/K8s service name) - database: e.g.,

mydb

- username: e.g.,

myuser

- password: stored securely by Prefect - Optional: - port: leave blank for default 5432; set it if your DB runs on a different port - query (dict): extra params, e.g.

{"sslmode":"require","connect_timeout":"10"}

Alternative (single-field option) - Switch the block to “Connection string” and enter one URL: -

<postgresql+psycopg2://user:pass@host:5432/db?sslmode=require>

Docs: prefect-sqlalchemy integration