PostgreSQL. For self‑hosted Prefect (open source), the only supported database backends are SQLite (default) and PostgreSQL. MySQL/MariaDB and others are not supported. How to use PostgreSQL: - Ensure the

pg_trgm

extension is installed/enabled in your database. - Set the connection URL via environment variable:

export PREFECT_SERVER_DATABASE_CONNECTION_URL="<postgresql+asyncpg://USER:PASSWORD@HOST:5432/DBNAME>"

- Typical format:

<postgresql+asyncpg://user:password@host:port/database>

Helpful docs: - Server overview and requirements (incl. pg_trgm): Prefect Server overview - Database config API reference: Database configurations - Settings reference (all DB env vars): Settings reference - Docker Compose example with Postgres: Self-hosted with Docker Compose

Short answer: control what gets written (logs/events/artifacts), set retention where available, prune old data periodically, and vacuum the DB to reclaim space. Practical steps for Prefect 3.x 1) Reduce what you store - Logs to API - Disable entirely if you don’t need logs in the UI:

export PREFECT_LOGGING_TO_API_ENABLED=false

- Or keep but reduce volume:

# Raise the level (fewer logs)
    export PREFECT_LOGGING_LEVEL=WARNING
    # Truncate very large lines (bytes)
    export PREFECT_LOGGING_TO_API_MAX_LOG_SIZE=20000

- Events - If you’re not using events/automations, disable persistence:

export PREFECT_SERVER_SERVICES_EVENT_PERSISTER_ENABLED=false

- Or shorten retention (ISO-8601 period; default is often P7D):

export PREFECT_SERVER_EVENTS_RETENTION_PERIOD=P3D

- Artifacts - Avoid writing large artifacts or only store keys/links, not big payloads. Settings reference: All settings and env vars 2) Prune old data (logs, runs, artifacts) - Delete logs for old runs without deleting the runs:

import asyncio
  from datetime import datetime, timedelta, timezone
  from prefect.client.orchestration import get_client
  from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime

  async def main(days=30, page_size=500):
      cutoff = datetime.now(timezone.utc) - timedelta(days=days)
      async with get_client() as client:
          page = 1
          while True:
              filters = FlowRunFilter(start_time=FlowRunFilterStartTime(before_=cutoff))
              runs = await client.read_flow_runs(flow_run_filter=filters, limit=page_size, offset=(page-1)*page_size)
              if not runs:
                  break
              for r in runs:
                  # removes only logs for this run
                  await client.delete_flow_run_logs(r.id)
              page += 1

  asyncio.run(main(days=30))

- Delete old runs entirely (cascades will remove related rows like logs; this reclaims more space):

import asyncio
  from datetime import datetime, timedelta, timezone
  from prefect.client.orchestration import get_client
  from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime

  async def main(days=90, page_size=200):
      cutoff = datetime.now(timezone.utc) - timedelta(days=days)
      async with get_client() as client:
          page = 1
          while True:
              filters = FlowRunFilter(start_time=FlowRunFilterStartTime(before_=cutoff))
              runs = await client.read_flow_runs(flow_run_filter=filters, limit=page_size, offset=(page-1)*page_size)
              if not runs:
                  break
              for r in runs:
                  await client.delete_flow_run(r.id)
              page += 1

  asyncio.run(main(days=90))

- Delete artifacts tied to old runs (if needed):

import asyncio
  from datetime import datetime, timedelta, timezone
  from prefect.client.orchestration import get_client
  from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterFlowRunId

  async def delete_artifacts_for_runs(run_ids):
      async with get_client() as client:
          for run_id in run_ids:
              af = ArtifactFilter(flow_run_id=ArtifactFilterFlowRunId(any_=[run_id]))
              arts = await client.read_artifacts(artifact_filter=af, limit=1000)
              for a in arts:
                  await client.delete_artifact(a.id)

Note: Always test cleanup in a non-prod environment first and take a backup/snapshot. 3) Reclaim disk space after deletes - SQLite: - Stop the server, then:

sqlite3 /path/to/prefect.db "VACUUM; PRAGMA wal_checkpoint(TRUNCATE); PRAGMA optimize;"

- PostgreSQL: - Ensure autovacuum is enabled and healthy. After large deletes, consider: ``` VACUUM (ANALYZE);

-- As a last resort during maintenance windows (locks table): VACUUM (FULL, ANALYZE); REINDEX DATABASE your_db;

- Monitor bloat and autovacuum via pg_stat views.

4) Monitor where growth comes from
- PostgreSQL top tables by size:

SELECT relname AS table, pg_total_relation_size(relid) AS bytes FROM pg_catalog.pg_statio_user_tables ORDER BY 2 DESC LIMIT 20;

- SQLite approximate table sizes:

sqlite3 /path/to/prefect.db "SELECT name, sum(pgsize) as bytes FROM dbstat GROUP BY name ORDER BY bytes DESC LIMIT 20;"

Common knobs to tune
- Events retention:

PREFECT_SERVER_EVENTS_RETENTION_PERIOD=P3D PREFECT_SERVER_SERVICES_EVENT_PERSISTER_ENABLED=false # if you do not need events

- Logs volume:

PREFECT_LOGGING_TO_API_ENABLED=false PREFECT_LOGGING_LEVEL=WARNING PREFECT_LOGGING_TO_API_MAX_LOG_SIZE=20000 ``` Docs to reference - Settings reference (env vars, including events/logging): Settings reference - Server overview: Server overview If you share: - Which backend (SQLite vs Postgres) - Current DB file/cluster size and growth per day - Whether you rely on events and artifacts - Desired retention windows for runs/logs/events …I can tailor a ready-to-run cleanup job and a minimal set of settings for your use case.

Short answer: mostly yes — the same principles apply in Prefect 2.x, but a few knobs are different. What carries over to 2.x - Databases: SQLite (default) and PostgreSQL are supported in OSS 2.x. Configure with:

export PREFECT_API_DATABASE_CONNECTION_URL="<postgresql+asyncpg://USER:PASSWORD@HOST:5432/DBNAME>"

- Reduce what you store: - Raise logging level to cut volume:

export PREFECT_LOGGING_LEVEL=WARNING

- You can disable sending logs to the API (check your exact 2.x version’s settings for the flag name; many 2.x versions support):

export PREFECT_LOGGING_TO_API_ENABLED=false

- Prune old data: - Delete logs for older runs or delete the runs entirely (will cascade logs/artifacts). The same client pattern generally works in 2.x:

import asyncio
    from datetime import datetime, timedelta, timezone
    from prefect.client.orchestration import get_client
    from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime

    async def main(days=90, page_size=200):
        cutoff = datetime.now(timezone.utc) - timedelta(days=days)
        async with get_client() as client:
            page = 1
            while True:
                filters = FlowRunFilter(start_time=FlowRunFilterStartTime(before_=cutoff))
                runs = await client.read_flow_runs(flow_run_filter=filters, limit=page_size, offset=(page-1)*page_size)
                if not runs:
                    break
                for r in runs:
                    # delete just logs:
                    # await client.delete_flow_run_logs(r.id)

                    # or delete the run (cascades):
                    await client.delete_flow_run(r.id)
                page += 1

    asyncio.run(main(days=90))

- Always test on non‑prod and back up first. - Reclaim space: - SQLite: stop server, then

VACUUM;

and checkpoint WAL. - PostgreSQL: rely on autovacuum; after large deletes run

VACUUM (ANALYZE);

or, during a maintenance window,

VACUUM FULL;

and consider

REINDEX

. Key differences from 3.x - Events: 2.x OSS does not persist the server-side events stream like 3.x. So 3.x-specific env vars such as

PREFECT_SERVER_SERVICES_EVENT_PERSISTER_ENABLED

and

PREFECT_SERVER_EVENTS_RETENTION_PERIOD

do not apply in 2.x OSS. - Settings names: database URL in 2.x is

PREFECT_API_DATABASE_CONNECTION_URL

(not

PREFECT_SERVER_DATABASE_CONNECTION_URL

). Some logging handler flags differ slightly by minor version; confirm in the 2.x settings reference below. Helpful 2.x docs - 2.x self-hosting guide: Self-hosted (2.x) - 2.x logging guide: Logging (2.x) - 2.x settings reference: Settings (2.x) - 2.x client usage: Using the client (2.x) If you share your exact Prefect 2.x version, I can map the precise env var names in that release and tailor a ready-to-run cleanup flow. Also, if you can upgrade to 3.x, you’ll gain better event controls and additional server settings.

Short answer: yes, you can prune directly in the DB, but the exact tables depend on your Prefect version. Can you confirm if you’re on Prefect 3.x or 2.x? The schema differs slightly. Below are safe targets and patterns for both versions. Always stop the Prefect server first, back up the DB, test on staging, then VACUUM/ANALYZE afterward. Prefect 3.x (self-hosted server) Common largest tables to prune: - log — flow/task run logs - event and event_resource — only if the event persister is enabled - artifact and artifact_collection — artifacts recorded by runs - flow_run_states and task_run_states — older/non-current states - flow_runs and task_runs — old runs in terminal states (largest reclaim) Cautious delete order - If you only want to reduce size without removing runs: - Delete old logs - Delete old events (and their event_resource rows) - Delete old artifacts - Optionally delete non-current state rows (do not remove any state referenced as the current state by a run) - If you want to purge old runs entirely (more space reclaimed), delete in this order: 1) task_run_states for task_runs you will delete 2) task_runs 3) flow_run_states that belong to flow_runs you will delete 4) artifacts associated with those runs (if any) 5) logs for those runs 6) flow_runs 7) artifact_collection entries now orphaned (if any) Note: Some FKs may cascade, but don’t rely on it; delete children first to avoid constraint errors. Example SQL (PostgreSQL; adjust cutoff) - Logs older than 30 days:

DELETE FROM log
WHERE timestamp < NOW() - INTERVAL '30 days';

- Events older than 7 days (when persisted):

DELETE FROM event_resource
WHERE event_id IN (
  SELECT id FROM event
  WHERE occurred < NOW() - INTERVAL '7 days'
);

DELETE FROM event
WHERE occurred < NOW() - INTERVAL '7 days';

- Artifacts older than 30 days:

DELETE FROM artifact
WHERE created < NOW() - INTERVAL '30 days';

- Non-current flow states older than 30 days (avoids removing current state_id used by flow_runs):

WITH current_state_ids AS (
  SELECT state_id FROM flow_run WHERE state_id IS NOT NULL
)
DELETE FROM flow_run_states s
WHERE s.id NOT IN (SELECT state_id FROM current_state_ids)
  AND s.timestamp < NOW() - INTERVAL '30 days';

- Delete old task runs in terminal states older than 90 days:

-- remove their states and logs first (example shows states)
DELETE FROM task_run_states s
USING task_runs tr
WHERE s.task_run_id = tr.id
  AND tr.end_time < NOW() - INTERVAL '90 days'
  AND tr.state_type IN ('COMPLETED', 'FAILED', 'CANCELLED');

DELETE FROM task_runs
WHERE end_time < NOW() - INTERVAL '90 days'
  AND state_type IN ('COMPLETED', 'FAILED', 'CANCELLED');

- Then do the same for flow runs:

DELETE FROM flow_run_states s
USING flow_runs fr
WHERE s.flow_run_id = fr.id
  AND fr.end_time < NOW() - INTERVAL '90 days'
  AND fr.state_type IN ('COMPLETED', 'FAILED', 'CANCELLED');

DELETE FROM flow_runs
WHERE end_time < NOW() - INTERVAL '90 days'
  AND state_type IN ('COMPLETED', 'FAILED', 'CANCELLED');

SQLite notes - Replace NOW() - INTERVAL with datetime('now','-30 days') syntax and column names that exist in your DB. - After large deletes:

VACUUM;
PRAGMA wal_checkpoint(TRUNCATE);
PRAGMA optimize;

Prefect 2.x (Orion) - No server-side event persistence in OSS 2.x, so typical large tables are: - log - flow_runs, task_runs - states (or flow_run_states/task_run_states depending on minor version) - artifacts (if you use them) - Use the same general strategy: - Delete old rows from log first - Delete artifacts if used - Delete non-current states (do not remove the current state referenced by runs) - Delete old task_runs first, then their flow_runs - Column names in 2.x can vary slightly by minor version (e.g., a single states table vs separate flow_run_states/task_run_states). Inspect schema before running deletes: - PostgreSQL:

\dt
    \d+ flow_runs

- SQLite: ```

.tables PRAGMA table_info(flow_runs);

Find the biggest tables first
- PostgreSQL:

SELECT relname AS table, pg_total_relation_size(relid) AS bytes FROM pg_catalog.pg_statio_user_tables ORDER BY 2 DESC LIMIT 20;

- SQLite:

SELECT name, SUM(pgsize) AS bytes FROM dbstat GROUP BY name ORDER BY bytes DESC LIMIT 20; ``` Important cautions - Stop the Prefect server while pruning to avoid race conditions. - Always leave the “current” state row for each run intact. - If you remove runs, remove their child rows first unless you’ve verified ON DELETE CASCADE exists. - Take a backup before you start and VACUUM/ANALYZE afterward. If you can share: - Prefect version (2.x vs 3.x) - DB engine (SQLite/Postgres) - Desired retention window (e.g., keep 30/90 days) …I’ll craft exact DELETE statements tailored to your schema and cutoff, including SQLite-compatible variants. Docs for reference - Prefect 3.x server overview: Server overview - Prefect 3.x settings (events/logging): Settings reference

Great question. In Prefect 3.x, setting

cache_expiration

enables task result caching; there isn’t a single “clear cache” button, but here are your options depending on what you want: - Bypass the cache for a run (don’t use the cached value this time) - Set

refresh_cache=True

on the task. This will force execution and write a fresh result, ignoring any existing cached state.

from datetime import timedelta
  from prefect import task

  @task(cache_expiration=timedelta(hours=1), refresh_cache=True)
  def my_task(...):
      ...

- Disable caching entirely going forward - Set a “no cache” policy; this ignores `cache_expiration`:

from prefect import task
  from prefect.cache_policies import NO_CACHE

  @task(cache_policy=NO_CACHE)
  def my_task(...):
      ...

- Alternatively, just remove

cache_expiration

(and any

cache_key_fn

) so caching isn’t used. - Invalidate existing caches proactively - Bump the task’s

version

so its cache key changes and existing cached results are not reused:

@task(version="2.0.0", cache_expiration=timedelta(hours=1))
  def my_task(...):
      ...

- Change your

cache_key_fn

to produce a different key (e.g., include a salt/env var) so old entries won’t match. - Physically remove persisted results - Prefect stores cached results in your configured result storage (local filesystem, S3, etc.). If you need to delete the actual saved artifacts, remove them from that storage location. Note: deleting files does not remove the “cached state” metadata in the Prefect database, but it will prevent old results from being read if referenced. - Remove cache state records - There’s no dedicated SDK/CLI to delete cached task states. You can delete the flow runs that created them (via UI or

prefect flow-run delete <id>

), which removes their states; if you also persist results, you may still need to clean the storage objects separately. Quick recap: - One-off bypass:

refresh_cache=True

- Turn off caching:

cache_policy=NO_CACHE

(or remove

cache_expiration

) - Invalidate keys: change

version

or

cache_key_fn

- Remove artifacts: delete objects in your result storage - There’s no “clear cache” API; expiration and key changes are the normal mechanisms What are you hoping to do right now: bypass for a single run, disable caching, or purge old results? I can tailor steps once I know which outcome you need.

In Prefect 3.x, you don’t pass arbitrary args directly to `on_completion`/`on_failure`. Hooks have fixed signatures and Prefect calls them for you: - Flow hooks:

(flow, flow_run, state)

- Task hooks:

(task, task_run, state)

To “pass variables” into a hook, use one of these patterns: 1) Bind extras with functools.partial or a closure - Works when you want to configure a static value (e.g., a Slack channel, feature flag) at definition time.

from functools import partial
from prefect import flow, task

def flow_notify(flow, flow_run, state, channel):
    # use your extra variable
    print(f"[{channel}] {flow_run.name} -> {state.type}")

@flow(on_completion=[partial(flow_notify, channel="#alerts")])
def my_flow(x: int):
    return x * 2

Or via a closure:

def make_flow_notify(channel):
    def hook(flow, flow_run, state):
        print(f"[{channel}] {flow_run.name} -> {state.type}")
    return hook

@flow(on_failure=[make_flow_notify("#incidents")])
def my_flow(x: int):
    ...

Same idea for tasks:

from functools import partial
from prefect import task

def task_done(task, task_run, state, threshold: int):
    if state.is_completed():
        print(f"Task {task_run.name} completed; threshold={threshold}")

@task(on_completion=[partial(task_done, threshold=5)])
def compute(x):
    return x + 1

2) Read values from the run objects provided to the hook (dynamic per-run) - Hooks receive rich context you can use to get parameters, tags, IDs, names, etc. Flow example (read flow parameters):

def flow_notify(flow, flow_run, state):
    env = flow_run.parameters.get("env", "unknown")
    print(f"Flow {flow_run.name} finished in env={env} with state={state.type}")

@flow(on_completion=[flow_notify])
def my_flow(env: str):
    ...

Task example (read task inputs/params):

def task_fail(task, task_run, state):
    # Inputs passed to the task are in task_run.parameters
    print(f"Task {task_run.name} failed; inputs={task_run.parameters}")

@task(on_failure=[task_fail])
def fragile(a, b):
    ...

Notes - Hooks can be sync or async; Prefect will await async hooks. - If you need other runtime details, you can also get them via

prefect.context.get_run_context()

inside the hook. - Hooks are set at decoration time; for per-run “variables,” prefer reading from

flow_run.parameters

or

task_run.parameters

(or close over environment/config values).

Great question! Assuming Prefect 3.x, here are the two supported ways to get the current flow/task name and state from any called function. - Simple names/ids: use the high-level runtime API - Current state objects: use the run context API Example that works from an arbitrary called function:

from prefect import flow, task, runtime
from prefect.context import get_run_context
from prefect.exceptions import MissingContextError

def log_where_am_i():
    # Names/IDs via the runtime API (safe to call anywhere)
    flow_name = runtime.flow_run.flow_name
    task_name = runtime.task_run.task_name
    flow_id = runtime.flow_run.id
    task_id = runtime.task_run.id

    print(f"Flow: {flow_name} (id={flow_id}), Task: {task_name} (id={task_id})")

    # State via the run context API
    try:
        ctx = get_run_context()  # returns TaskRunContext or FlowRunContext
    except MissingContextError:
        print("Not running inside a Prefect flow/task.")
        return

    if hasattr(ctx, "task_run"):
        # We're inside a task
        print(f"Task state: {ctx.task_run.state.name} ({ctx.task_run.state.type})")
    else:
        # We're inside a flow
        print(f"Flow state: {ctx.flow_run.state.name} ({ctx.flow_run.state.type})")

@task
def my_task():
    log_where_am_i()

@flow
def my_flow():
    log_where_am_i()
    my_task()

if __name__ == "__main__":
    my_flow()

Notes: -

prefect.runtime

is the easiest way to get names and IDs; it won’t raise if you call it outside a flow/task and just returns empty values. - To get the current state, use

prefect.context.get_run_context()

(raises

MissingContextError

outside Prefect). If you prefer no exceptions, you can use

TaskRunContext.get()

/

FlowRunContext.get()

which return

None

when unavailable. Helpful references: - Runtime context docs: Runtime context (Prefect 3) - Source: prefect.runtime.flow_run, prefect.runtime.task_run, prefect.context If you’re on Prefect 2.x instead, let me know and I’ll adjust the examples.

Short answer: you don’t need to install Prefect via EXTRA_PIP_PACKAGES if you use the official Prefect Docker image. You do need to ensure any extra integrations (like prefect-shell) are available in the flow-run container. Guidance: - Using Prefect’s base image (default when you let .deploy() build the image): Prefect is already installed. Just add your extra libs (e.g., prefect-shell) to requirements.txt so they’re baked into the image. - Using your own base image (e.g., python:3.11-slim): you must install prefect (and your extras) in the image at build time, or use EXTRA_PIP_PACKAGES to install them at container start. - EXTRA_PIP_PACKAGES works only with the Prefect base image’s entrypoint. It’s fine for quick tests, but slower and less reproducible than baking deps into the image. Recommended (build-time): - requirements.txt

prefect-shell>=0.2.0
# any other libs you need

- Let .deploy() build from a Prefect image, or use a custom Dockerfile that starts FROM prefecthq/prefect:3-latest and installs requirements.txt. If you must install at runtime (not recommended for prod), set it on the deployment:

my_flow.deploy(
  name="my-deployment",
  work_pool_name="my-docker-pool",
  image="prefecthq/prefect:3-latest",
  build=False,
  job_variables={"env": {"EXTRA_PIP_PACKAGES": "prefect-shell"}}
)

Notes: -

from prefect.blocks.system import JSON

comes from core Prefect (already in the Prefect image). -

from prefect_shell import ShellOperation

requires the prefect-shell package. Docs: - Docker work pools - Deployments If you can share which image you plan to run, I can suggest the minimal Dockerfile or requirements setup.

Assuming Prefect 3.x, you can update an existing SqlAlchemyConnector block by loading it, changing its connection info, and saving it back with overwrite=True. Method A: Python (load ➜ mutate ➜ save)

# pip install prefect-sqlalchemy

from prefect_sqlalchemy import SqlAlchemyConnector
from prefect_sqlalchemy.credentials import ConnectionComponents, SyncDriver

# Load the existing block by name
connector = SqlAlchemyConnector.load("my-sqlalchemy-block")

# Option 1: set a new SQLAlchemy URL (works for any DB)
connector.connection_info = "<postgresql+psycopg2://user:pass@new-host.example.com:5432/new_database>"
# For SQL Server (example):
# connector.connection_info = "<mssql+pyodbc://user:pass@new-server/new_db?driver=ODBC+Driver+18+for+SQL+Server>"

connector.save("my-sqlalchemy-block", overwrite=True)

Or, if you prefer structured components:

from prefect_sqlalchemy import SqlAlchemyConnector
from prefect_sqlalchemy.credentials import ConnectionComponents, SyncDriver

connector = SqlAlchemyConnector.load("my-sqlalchemy-block")

connector.connection_info = ConnectionComponents(
    driver=SyncDriver.POSTGRESQL_PSYCOPG2,  # pick the appropriate driver
    host="<http://new-host.example.com|new-host.example.com>",
    port=5432,
    database="new_database",
    username="user",
    password="pass",
)

connector.save("my-sqlalchemy-block", overwrite=True)

Method B: UI - Navigate to Blocks in the Prefect UI - Open your SQLAlchemy Connector block - Click Edit, update Host/Server and Database, and Save Notes: - Use overwrite=True to update the existing block in place. - The change will be used by subsequent flow runs that load this block. - Make sure prefect-sqlalchemy is installed where you run the update script. - Repo reference: prefect-sqlalchemy

Short answer: - Yes, every deployment needs an entrypoint (file+function or module path) so Prefect knows which flow to load. - No, you do not need to copy your files to the worker host if your flow code is already baked into the Docker image used for the flow run. How it fits together in Prefect 3.x: - The deployment stores an entrypoint like path/to/file.py:flow_func (FILE_PATH) or package.module:flow_func (MODULE_PATH). This tells the worker how to import your flow at runtime. - For Docker work pools, the worker launches a container using your image. If your image already contains the project files, Prefect will import the flow from inside the container using that entrypoint. The worker host does not need the code. - If you use the Python API my_flow.deploy(...), Prefect will auto-compute the entrypoint from the flow function so you usually don’t need to type it. - If you use prefect deploy (CLI) or prefect.yaml, specify the entrypoint explicitly. Tips to avoid path issues: - Make sure the entrypoint path matches the location of the code inside the image. If your Dockerfile copies the project to /opt/prefect/app, either: - Use FILE_PATH relative to the container’s working directory, or - Use MODULE_PATH if your code is on PYTHONPATH. - Example FILE_PATH: flows/my_flow.py:my_flow - Example MODULE_PATH: my_package.flows.my_flow:my_flow You only need local files on the worker if you are not baking code into the image and you are not pulling it from remote storage; otherwise, baked-in or pulled code is sufficient.

Here’s a concise, working example of a Docker-based deployment in Prefect 3.x. Quick start steps 1) Install requirements (Docker must be running)

pip install "prefect>=3" prefect-docker

2) Create a Docker work pool (once)

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

3) Start a worker for that pool (in a separate terminal) - If you installed with pip:

prefect worker start --pool docker-pool

- Or ensure the docker integration is available via uv:

uv run --with prefect-docker prefect worker start --pool docker-pool

4) Define your flow (my_flow.py)

from prefect import flow

@flow(log_prints=True)
def hello(name: str = "world"):
    print(f"Hello, {name}!")

5) Programmatically create a Docker deployment (builds a local image automatically)

from prefect import flow

@flow(log_prints=True)
def hello(name: str = "world"):
    print(f"Hello, {name}!")

if __name__ == "__main__":
    # This will:
    # - build a Docker image from your project (auto-generated Dockerfile)
    # - register a deployment that uses the "docker-pool" work pool
    hello.deploy(
        name="hello-docker",
        work_pool_name="docker-pool",
        image="hello-image:latest",  # local image name
        build=True,                  # build image from your project
        push=False,                  # skip push for local dev
        job_variables={
            "env": {"EXTRA_PIP_PACKAGES": "pandas"},   # add packages at runtime
            "volumes": ["/host/data:/data:ro"],        # optional volume mount
            # other useful options: mem_limit, auto_remove, stream_output, etc.
        },
        parameters={"name": "Prefect"},
        tags=["example", "docker"],
        description="Simple Docker deployment example",
    )

6) Kick off a run of the deployment

prefect deployment run "hello/hello-docker" --param name=Marvin

Alternative: CLI with prefect.yaml prefect.yaml (in your project root)

deployments:
  - name: hello-docker
    entrypoint: my_flow.py:hello
    work_pool:
      name: docker-pool
      job_variables:
        env:
          EXTRA_PIP_PACKAGES: "pandas"
        volumes:
          - "/host/data:/data:ro"
    parameters:
      name: "world"

Deploy from the file:

prefect deploy -n hello-docker

Notes - In Prefect 3.x, use

flow.deploy(...)

or

flow.from_source(...).deploy(...)

. Do not use the old 2.x methods like

Deployment.build_from_flow()

or

prefect deployment build

. - Common Docker job_variables you can set per deployment include:

image

,

image_pull_policy

,

env

,

volumes

,

networks

,

network_mode

,

auto_remove

,

stream_output

,

mem_limit

,

memswap_limit

,

privileged

, and

container_create_kwargs

. If you want a remote-code example (deploy directly from a Git repo), I can share that too.

Short answer: yes — when you run

prefect deploy

the CLI must be able to import your flow from the local project on the machine doing the deploy. An entrypoint like

/app/my_flow.py:start_flow

points to a path inside your container, not on your local filesystem, so the CLI can’t load it and you’ll see “Failed to load flow from entrypoint …”. How to fix it - Use a project-relative entrypoint (not a container path), and run

prefect deploy

from the project root where that file exists. - Keep the same relative layout inside your Docker image so the entrypoint also resolves at runtime in the container. Example prefect.yaml

deployments:
  - name: start-flow-docker
    entrypoint: my_flow.py:start_flow   # relative to your local project
    work_pool:
      name: docker-pool
      job_variables:
        image: "your-repo/your-image:latest"
        # optional: env/volumes/etc.

Example Dockerfile (make the container match the project layout)

FROM prefecthq/prefect:3-latest
WORKDIR /app
COPY . .
# optional: pip install -r requirements.txt
# Now, inside the container the file is at /app/my_flow.py,
# but the deployment entrypoint remains "my_flow.py:start_flow"

Then:

prefect work-pool create --type docker docker-pool
prefect worker start --pool docker-pool     # separate terminal
prefect deploy -n start-flow-docker         # run from the project root containing my_flow.py

If you cannot have the code locally on the deploy machine - Use a programmatic deployment that pulls code from a remote source:

from prefect import flow

flow.from_source(
    source="<https://github.com/your-org/your-repo.git>",
    entrypoint="path/in/repo/my_flow.py:start_flow",
).deploy(
    name="start-flow-docker",
    work_pool_name="docker-pool",
    image="your-repo/your-image:latest",
)

Quick checklist - entrypoint is relative (e.g.,

my_flow.py:start_flow

), not absolute

/app/...

- you run

prefect deploy

from the folder where

my_flow.py

exists - your Dockerfile copies the same project into the image and sets a WORKDIR so the same relative entrypoint works inside the container If you can share your

prefect.yaml

entrypoint and the relevant part of your Dockerfile, I can pinpoint the minimal change.

Thanks for the details — the error is because your entrypoint is pointing to an absolute container path (

/app/...

). With

prefect deploy

, the entrypoint must be a path relative to the repository root (not the Docker WORKDIR), and when you use a

pull

step (git clone), the worker will clone your repo at run time and load the entrypoint from there. What to change - In prefect.yaml, set

entrypoint

to the repo-relative path (e.g.,

path/in/repo/my_flow.py:start_flow

), not

/app/my_flow.py:start_flow

. - Keep your

Dockerfile

WORKDIR as

/app

if you like, but it does not affect the entrypoint in prefect.yaml. Minimal prefect.yaml for “pull from GitHub + existing Docker image”

name: my-project

pull:
  - prefect.deployments.steps.git_clone:
      repository: <https://github.com/your-org/your-repo.git>
      branch: main
      # for private repos:
      # credentials: "{{ prefect.blocks.github-credentials.my-creds }}"

deployments:
  - name: start-flow-docker
    entrypoint: path/in/repo/my_flow.py:start_flow  # relative to repo root
    work_pool:
      name: docker-pool
      job_variables:
        image: your-registry/your-image:tag   # use your prebuilt image
        # optional: env/volumes/mem_limit/etc.

Then run:

prefect work-pool create --type docker docker-pool   # once
prefect worker start --pool docker-pool              # in a separate terminal
prefect deploy -n start-flow-docker                  # from the folder with prefect.yaml

If you still see “Failed to load flow from entrypoint …” - Easiest: temporarily clone the repo locally so the CLI can import the flow when creating the deployment: -

git clone <https://github.com/your-org/your-repo.git>

-

cd your-repo

(so

path/in/repo/my_flow.py

exists locally) -

prefect deploy -n start-flow-docker

- Or, switch to a programmatic deployment that pulls from source (does not require local code at deploy time):

from prefect import flow

flow.from_source(
    source="<https://github.com/your-org/your-repo.git>",
    entrypoint="path/in/repo/my_flow.py:start_flow",
).deploy(
    name="start-flow-docker",
    work_pool_name="docker-pool",
    image="your-registry/your-image:tag",
)

If you paste your current

prefect.yaml

(pull section + deployments entrypoint), I can spot the exact tweak needed.

You’ve got two good options for authenticating to a private GitHub repo with Prefect 3.x: Option A — Use a GitHubCredentials block (recommended) 1) Install/register the integration

pip install "prefect[github]"
prefect block register -m prefect_github

2) Create the block - Via Python:

from prefect_github import GitHubCredentials

GitHubCredentials(token="ghp_...").save(
  "my-github-creds",
  overwrite=True
)

- Or via the Prefect UI: Blocks > + > GitHub Credentials > paste PAT > save as

my-github-creds

3) Reference it in prefect.yaml pull step

pull:
  - prefect.deployments.steps.git_clone:
      repository: <https://github.com/your-org/your-private-repo.git>
      branch: main
      credentials: "{{ prefect.blocks.github-credentials.my-github-creds }}"

Option B — Use a Secret block for the token 1) Create a Secret block - Python:

from prefect.blocks.system import Secret
Secret(value="ghp_...").save("github-pat", overwrite=True)

- Or via UI: Blocks > + > Secret > value: your PAT > name:

github-pat

2) Reference it with access_token (don’t combine with credentials)

pull:
  - prefect.deployments.steps.git_clone:
      repository: <https://github.com/your-org/your-private-repo.git>
      branch: main
      access_token: "{{ prefect.blocks.secret.github-pat }}"

Notes and tips - Use either

credentials

OR

access_token

(using both will error). - Entrypoint should be repo-relative, e.g.

path/in/repo/flow.py:my_flow

. - For programmatic deployments, pass credentials via a GitRepository:

from prefect import flow
from prefect.runner.storage import GitRepository
from prefect_github import GitHubCredentials

repo = GitRepository(
  url="<https://github.com/your-org/your-private-repo.git>",
  credentials=GitHubCredentials.load("my-github-creds"),
)

flow.from_source(
  source=repo,
  entrypoint="path/in/repo/flow.py:my_flow",
).deploy(
  name="my-deployment",
  work_pool_name="docker-pool",
)

PAT scopes - Use a fine‑grained PAT with at least: - Repository access: your target repo(s) - Permissions: Contents (Read), Metadata (Read) If you share your exact

pull:

section, I can validate the block reference for you.

Short answer - If you’re using a prebuilt Docker image that already contains your flow code and you use

prefect deploy

or

flow.deploy(...)

, you still need a local copy of the entrypoint file so Prefect can import the flow at deploy time. - The only way to avoid having the code locally is to deploy “from source” (e.g., GitHub) using a pull step or

flow.from_source(...)

. You can still run the deployment with your prebuilt image, but the worker will clone the repo at runtime to import the flow. Why - The deploy step loads your flow to read metadata (name, parameter schema, etc.). Without a local file (or a remote source), Prefect can’t import it and will error. Option 1 — Baked image only (no pull), keep a tiny local copy - Keep a minimal local copy of the entrypoint file with the same function name and path as inside your image. - Ensure the relative path matches both locally and inside the container (WORKDIR /app is fine). prefect.yaml

name: my-project
pull: []  # no git clone; rely on baked image

deployments:
  - name: baked-image-deployment
    entrypoint: app/my_flow.py:start_flow   # relative to your local project
    work_pool:
      name: docker-pool
      job_variables:
        image: your-registry/your-image:tag
        # optional: env/volumes/mem_limit, etc.

Dockerfile (example)

FROM your-base-image
WORKDIR /app
COPY . .   # this image already contains your code
# pip install -r requirements.txt (if needed)

Run

prefect work-pool create --type docker docker-pool   # once
prefect worker start --pool docker-pool              # separate terminal
prefect deploy -n baked-image-deployment             # run where app/my_flow.py exists locally

Option 2 — No local code: deploy from GitHub, still use your prebuilt image - Worker will clone the repo at runtime to import the flow, but container image is still the one you built. prefect.yaml

name: my-project

pull:
  - prefect.deployments.steps.git_clone:
      repository: <https://github.com/your-org/your-repo.git>
      branch: main
      # credentials: "{{ prefect.blocks.github-credentials.my-github-creds }}"  # for private repo

deployments:
  - name: baked-image-deployment
    entrypoint: path/in/repo/app/my_flow.py:start_flow
    work_pool:
      name: docker-pool
      job_variables:
        image: your-registry/your-image:tag

Or programmatic

from prefect import flow
from prefect.runner.storage import GitRepository
# from prefect_github import GitHubCredentials

repo = GitRepository(
    url="<https://github.com/your-org/your-repo.git>",
    # credentials=GitHubCredentials.load("my-github-creds"),
)

flow.from_source(
    source=repo,
    entrypoint="path/in/repo/app/my_flow.py:start_flow",
).deploy(
    name="baked-image-deployment",
    work_pool_name="docker-pool",
    image="your-registry/your-image:tag",
)

Practical tip - If you want truly “no remote pull” at runtime, keep a tiny local copy of the entrypoint file for deploy time (it can be the exact same code as in your image, just minimal). Prefect will register the deployment from that local file; at run time the worker will use your prebuilt image and the entrypoint path will resolve inside the container. If you share your entrypoint path and image’s WORKDIR/layout, I can suggest the minimal local file structure to make this work.

You can list blocks from your Prefect server using the Python client. Here are the most common patterns. Async (recommended)

import asyncio
from prefect.client.orchestration import get_client

async def list_all_blocks(limit=200):
    async with get_client() as client:
        blocks = await client.read_block_documents(limit=limit, include_secrets=False)
        for b in blocks:
            type_slug = getattr(getattr(b, "block_type", None), "slug", None)
            print(f"- {b.name} | type={type_slug or b.block_type_name} | id={b.id} | created={b.created}")

asyncio.run(list_all_blocks())

List blocks by type

import asyncio
from prefect.client.orchestration import get_client

async def list_blocks_by_type(type_slug: str, limit=100):
    async with get_client() as client:
        docs = await client.read_block_documents_by_type(
            block_type_slug=type_slug, limit=limit, include_secrets=False
        )
        for d in docs:
            print(f"{d.name} -> {d.data}")

asyncio.run(list_blocks_by_type("secret"))

Get one block by name and type

import asyncio
from prefect.client.orchestration import get_client

async def get_block(type_slug: str, name: str):
    async with get_client() as client:
        doc = await client.read_block_document_by_name(
            block_type_slug=type_slug, name=name, include_secrets=False
        )
        print(doc)

asyncio.run(get_block("github-credentials", "my-github-creds"))

Sync usage (if you prefer)

from prefect.client.orchestration import get_client

with get_client(sync_client=True) as client:
    docs = client.read_block_documents(limit=100, include_secrets=False)
    for d in docs:
        print(d.name, d.block_type_name, d.id)

Notes -

include_secrets=False

is safer when listing; set to True only when you actually need secret values. - By default, anonymous blocks are excluded. If you need everything, you can use the filter classes in

prefect.client.schemas.filters

to include anonymous ones or filter by name/ID. - Useful type slugs include

secret

,

github-credentials

,

docker-registry-credentials

,

s3-bucket

, etc. If you need to enumerate available block types:

import asyncio
from prefect.client.orchestration import get_client

async def list_block_types():
    async with get_client() as client:
        types = await client.read_block_types()
        for t in types:
            print(f"{t.name} (slug={t.slug})")

asyncio.run(list_block_types())

If you share whether you want all blocks across all types, or only named (non-anonymous) blocks of a particular integration, I can tailor the exact query/filter.