Great question — and nice job planning the upgrade. Since you’re on Prefect 2.20.16: 1) What version to update to - Recommend upgrading to the latest stable 3.x release and pinning to that exact patch for production. Today, that’s 3.6.6. See releases: Prefect releases. - If you self-host Prefect Server, plan to upgrade the server in lockstep and run the DB migration after the package upgrade. 2) Key things to consider for the transition Here are the most important changes you’ll hit moving from 2.20.x → 3.x, plus the docs to reference. - Deployments and CLI - Removed:

prefect deployment build/apply

and

Deployment.build_from_flow()

- New:

prefect deploy

and

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

or

flow.deploy(...)

- Deployment YAML moves to

prefect.yaml

at your repo root - Docs: Deploy via Python, Define deployments with prefect.yaml - Migration guide: Upgrade to Prefect 3 - Agents → Workers and Work Pools - Agents are replaced by workers, and “work queues” are now typed “work pools” - Start a worker for your pool (check

prefect worker start --help

) - If you had infrastructure blocks, publish them as work pools or configure a pool in the UI - Docs: Workers, Work pools, migration: Upgrade agents to workers - Storage changes - GitHub storage block from core is removed; use

.from_source('<https://github.com/...>')

with

GitRepository

(or a block from a collection), not a GitHub storage block - Cloud stores moved to collections (prefect-aws/gcp/azure). Update those packages alongside the core upgrade - Docs: Use code from Git, Persist workflow results - Scheduling -

schedule=

becomes

schedules=[...]

(a list). You can also pass

interval=

,

cron=

, or

rrule=

to

deploy/serve

- New schedule classes:

prefect.schedules.Interval

,

Cron

,

RRule

- Docs: Schedule flow runs, Create schedules - Flow/task behavior changes you may notice - Failure propagation: 3.x tightens how failures propagate. Don’t rely on “a task failed somewhere so the flow will auto-fail.” If you call tasks without handling, let exceptions propagate or explicitly check

return_state=True

and raise on failure. - Async tasks: call async tasks from async flows (i.e.,

@flow(async)

and

await my_task()

). - Caching: 3.x introduces a new cache policy system. If a task has side effects, set

cache_policy=None

on that task. - Result persistence: results are not persisted by default; enable per-flow/task or globally if needed. - Docs: Caching, Persist workflow results - Settings and configuration - 3.x supports env vars (

PREFECT_*

),

.env

,

prefect.toml

, or

[tool.prefect]

in

pyproject.toml

, plus profiles - Useful settings:

PREFECT_API_URL

,

PREFECT_API_KEY

,

PREFECT_RESULTS_PERSIST_BY_DEFAULT

,

PREFECT_DEFAULT_RESULT_STORAGE_BLOCK

- Docs: Settings & profiles, Settings reference - Pydantic v2 - Prefect moved to Pydantic v2. If you have your own Pydantic models, scan for v1→v2 breaking changes - Pydantic migration: Pydantic v2 migration

- Prefect Server (self-hosted) upgrade - Upgrade the Prefect package, then upgrade the DB schema before resuming schedules - Pause schedules during the cutover to avoid duplication on the first scheduler loop - Server docs: Self-host overview - Upgrade guide covers these steps too: Upgrade to Prefect 3 Minimal examples to migrate your deployments - Python API

from prefect import flow
from prefect.schedules import Interval
from datetime import timedelta

@flow
def my_flow():
    ...

# If your code lives in a repo (recommended)
my_flow.from_source(
    source="<https://github.com/your-org/your-repo>",  # or a GitRepository object
    entrypoint="path/to/file.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="your-pool",
    schedules=[Interval(timedelta(minutes=10))],
    parameters={"foo": "bar"},
)

- CLI workflow

# from your project root
prefect deploy
# Follow interactive prompts to create or update prefect.yaml and your deployment(s)

Suggested migration checklist - Pause deployment schedules - If self-hosting, snapshot your DB - Upgrade: pin to the latest 3.x (e.g., 3.6.6), and upgrade any collections you use (

prefect-aws

,

prefect-gcp

,

prefect-azure

, etc.) - Replace any use of

Deployment.build_from_flow

and

prefect deployment build/apply

with

flow.deploy(...)

or

prefect deploy

- Migrate agents → workers and create a typed work pool for your infrastructure - Convert

schedule=

to

schedules=[...]

(or use interval/cron/rrule params) - Review tasks for caching and result persistence needs - Test in staging, then resume schedules Handy links - Upgrade to Prefect 3: Migration guide - Agents → Workers: Agents to workers - Deployments: Deploy via Python, prefect.yaml - Schedules: Schedules - Results & caching: Results, Caching - Settings: Settings & profiles - Releases: Prefect releases A couple of clarifying questions so I can tailor exact steps and examples: - Are you on Prefect Cloud or self-hosted Server? - What infrastructure do you use for execution (local process, Docker, Kubernetes, ECS, etc.)? - Do you currently use agents,

prefect deployment build/apply

, or core storage blocks like GitHub/S3/GCS from

prefect.filesystems

? If so, which ones?

Short answer: upgrade your prefect-sqlalchemy. Version 0.4.0 and newer support SQLAlchemy 2.x while remaining compatible with 1.4, so you can move to Prefect 3.x without the pinning conflict. What to do - If you want to move to Prefect 3.x and SQLAlchemy 2.x: - Bump prefect-sqlalchemy to >=0.4.0 - Bump SQLAlchemy to >=2,<3 - Upgrade Prefect to 3.x Example (Poetry):

prefect = "^3.0.0"
prefect-sqlalchemy = "^0.4.0"
SQLAlchemy = ">=2,<3"

Example (pip):

pip install "prefect>=3" "prefect-sqlalchemy>=0.4.0" "SQLAlchemy>=2,<3"

- If you’d like to upgrade Prefect to 3.x but keep SQLAlchemy 1.4 for now: - You can still upgrade prefect-sqlalchemy to >=0.4.0 (it supports both 1.4 and 2.x) - Keep

SQLAlchemy = "^1.4"

until you’re ready to move to 2.x Notes - Prefect itself doesn’t require SQLAlchemy; the conflict is only from prefect-sqlalchemy 0.3.1 pinning SQLAlchemy <2. Upgrading to 0.4.0+ resolves it. - In Prefect 3.x you can also use the extras to let Prefect manage the integration:

pip install "prefect[sqlalchemy]>=3"

This installs the SQLAlchemy integration with the right constraints. - No API-breaking changes were introduced for typical usage of

SqlAlchemyConnector

when moving from 0.3.x to 0.4.x. References - Release adding SQLAlchemy 2.x support in prefect-sqlalchemy 0.4.0: Release notes - PR introducing SQLAlchemy 2.x compatibility: PR #70 - Prefect SQLAlchemy integration docs: Docs Want me to suggest the exact changes for your pyproject.toml or requirements.txt? If you share your current file, I’ll propose a minimal diff.

That error means you’re on Pydantic v2 (required by Prefect 3) but something in your stack is using a Pydantic v1 hook. With Prefect 3, the usual culprit is an older prefect-sqlalchemy (e.g., 0.3.1) built against Pydantic v1. Fix - Move to the Prefect 3–compatible line of the integration (prefect-sqlalchemy 0.5.x) and keep SQLAlchemy <3. - Easiest: install the Prefect extra so versions line up automatically. Commands - Pip (recommended):

pip install -U "prefect[sqlalchemy]" "SQLAlchemy>=2,<3"

or explicitly:

pip install -U "prefect>=3" "prefect-sqlalchemy>=0.5.0" "SQLAlchemy>=2,<3"

- Poetry:

prefect = "^3"
  prefect-sqlalchemy = "^0.5.0"
  SQLAlchemy = ">=2,<3"

Verify your environment

python -c "import prefect, pydantic, sqlalchemy, prefect_sqlalchemy as ps; print('prefect', prefect.__version__); print('pydantic', pydantic.__version__); print('sqlalchemy', sqlalchemy.__version__); print('prefect-sqlalchemy', getattr(ps, '__version__', 'unknown'))"

Why this works - Prefect 3 depends on Pydantic v2. Older prefect-sqlalchemy (e.g., 0.3.1) uses v1 internals like

__modify_schema__

, which triggers the error under v2. - The 0.5.x line of prefect-sqlalchemy is built for Prefect 3/Pydantic v2 and supports SQLAlchemy 1.4.x and 2.x (constraint is

<3

). References - Prefect install and integrations: https://docs.prefect.io/v3/how-to-guides/integrations/use-integrations/ - Prefect’s sqlalchemy extra (uses prefect-sqlalchemy >= 0.5.0): https://github.com/PrefectHQ/prefect/blob/3.0.0/pyproject.toml#L110-L130 - PR adding SQLAlchemy 2.x support in the integration: https://github.com/PrefectHQ/prefect-sqlalchemy/pull/70 If you paste your current dependency constraints (pyproject/requirements), I can suggest an exact minimal diff.

You’re hitting this because

DatabaseCredentials

was removed in the Prefect 3-compatible line of prefect-sqlalchemy (0.5.x). The replacement is to provide connection info directly to

SqlAlchemyConnector

via either: - a database URL string, or - a

ConnectionComponents

model (plus `SyncDriver`/`AsyncDriver` enums) How to fix your imports and code Option A — simplest (URL string)

from prefect_sqlalchemy.database import SqlAlchemyConnector

connector = SqlAlchemyConnector(
    connection_info="<postgresql+psycopg2://user:pass@host:5432/mydb>"
)

Option B — structured components

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

connector = SqlAlchemyConnector(
    connection_info=ConnectionComponents(
        driver=SyncDriver.POSTGRESQL_PSYCOPG2,
        username="user",
        password="pass",
        host="host",
        port=5432,
        database="mydb",
    )
)

Typical migration (old → new) - Old (0.3.x/0.4.x):

from prefect_sqlalchemy import DatabaseCredentials
from prefect_sqlalchemy.database import SqlAlchemyConnector

creds = DatabaseCredentials(
    username="user", password="pass", host="host", port=5432,
    database="mydb", driver="postgresql+psycopg2"
)
connector = SqlAlchemyConnector(credentials=creds)

- New (0.5.x):

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

connector = SqlAlchemyConnector(
    connection_info=ConnectionComponents(
        driver=SyncDriver.POSTGRESQL_PSYCOPG2,
        username="user", password="pass", host="host", port=5432, database="mydb",
    )
)

or

from prefect_sqlalchemy.database import SqlAlchemyConnector

connector = SqlAlchemyConnector(
    connection_info="<postgresql+psycopg2://user:pass@host:5432/mydb>"
)

Notes - In 0.5.x, don’t import `DatabaseCredentials`—it doesn’t exist. Use

ConnectionComponents

or a URL string. - Keep SQLAlchemy pinned to

<3

(2.x is fine). - With Prefect 3, prefer installing the extra so versions align:

pip install -U "prefect[sqlalchemy]" "SQLAlchemy>=2,<3"

Docs/refs - Integration docs: Prefect SQLAlchemy integration - Source layout (Prefect 3): prefect-sqlalchemy in Prefect repo If you share the snippet where you import/use

DatabaseCredentials

, I can give you a minimal diff to update it.

Great question—when you’re using blocks, the migration is a little different. In Prefect 3 / prefect-sqlalchemy 0.5.x, the old

DatabaseCredentials

block type was removed. The current block is

SqlAlchemyConnector

, and you should load that instead. You’ll need to migrate any saved

DatabaseCredentials

blocks to the new block type once. What to use now - Load the new block type:

from prefect_sqlalchemy.database import SqlAlchemyConnector

  connector = SqlAlchemyConnector.load(database_block_name)

- Then use it as a context manager or to get an engine/connection:

with connector:
      rows = connector.fetch_all("SELECT 1")
  # or
  engine = connector.get_engine()

How to migrate existing saved blocks (one-time) If you have existing saved

DatabaseCredentials

blocks, create and save a

SqlAlchemyConnector

with the same connection info. Example migration snippet:

# 1) Load the old block
from prefect_sqlalchemy import DatabaseCredentials  # only for migration in environments that still have it
old = DatabaseCredentials.load("my-old-db-creds")

# 2) Create the new connector
from prefect_sqlalchemy.database import SqlAlchemyConnector
from prefect_sqlalchemy.credentials import ConnectionComponents, SyncDriver

if getattr(old, "url", None):
    # If your old block used a full URL
    connector = SqlAlchemyConnector(
        connection_info=str(old.url),
        connect_args=getattr(old, "connect_args", None),
    )
else:
    # If your old block used individual fields
    connector = SqlAlchemyConnector(
        connection_info=ConnectionComponents(
            driver=SyncDriver(old.driver) if hasattr(SyncDriver, old.driver.name if hasattr(old.driver, "name") else "") else old.driver,
            username=old.username,
            password=old.password,
            host=old.host,
            port=old.port,
            database=old.database,
            query=getattr(old, "query", None),
        ),
        connect_args=getattr(old, "connect_args", None),
    )

# 3) Save the new block (you can reuse the same name)
connector.save("my-old-db-creds", overwrite=True)

If your environment no longer provides

DatabaseCredentials

(import error), you can’t load it directly to read the fields. In that case: - Recreate the new block from the original connection details you used when creating the old block (URL or components), or - If you have access to the Prefect server API and the workspace where the block is stored, we can query the block document JSON and reconstruct it. I can help with that if you share your workspace setup (Cloud or OSS) and the block name. Summary - Replace

DatabaseCredentials.load(name)

with

SqlAlchemyConnector.load(name)

going forward. - Migrate existing blocks once by saving a

SqlAlchemyConnector

with the same connection info. - Keep SQLAlchemy pinned to <3 (2.x is fine). Using

pip install -U "prefect[sqlalchemy]"

aligns all versions for Prefect 3. If you paste your current code that creates/loads the block and your DB type (Postgres/MySQL/etc.), I’ll send back an exact minimal diff for your project.

This is a SQLAlchemy 2.0 change:

Engine.execute(...)

was removed. In 2.x you must execute on a Connection (or use the higher-level Prefect helpers). The

OptionEngine

you see usually comes from calling

engine.execution_options(...)

, but even a plain

Engine

no longer has

execute

. Two easy fixes: Option A — use the Prefect connector methods (simplest)

from prefect_sqlalchemy.database import SqlAlchemyConnector

connector = SqlAlchemyConnector.load(database_block_name)

with connector:
    # single statement
    connector.execute("INSERT INTO t (id) VALUES (:id)", {"id": 1})

    # multiple rows
    connector.execute_many(
        "INSERT INTO t (id, name) VALUES (:id, :name)",
        [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}],
    )

    rows = connector.fetch_all("SELECT * FROM t")

Option B — use SQLAlchemy 2.0 style with a Connection

from sqlalchemy import text
from prefect_sqlalchemy.database import SqlAlchemyConnector

connector = SqlAlchemyConnector.load(database_block_name)
engine = connector.get_engine()

# sync driver
with engine.begin() as conn:
    conn.execute(text("INSERT INTO t (id) VALUES (:id)"), {"id": 1})
    res = conn.execute(text("SELECT * FROM t"))
    rows = res.fetchall()

# if using an async driver:
# async with engine.begin() as conn:
#     await conn.execute(text("..."))

Common test migration - Old (1.4 style):

engine = creds.get_engine()
  engine.execute("INSERT ...")

- New (2.0 style):

with SqlAlchemyConnector.load(name) as db:
      db.execute("INSERT ...")
  # or
  engine = SqlAlchemyConnector.load(name).get_engine()
  with engine.begin() as conn:
      conn.execute(text("INSERT ..."))

If your upsert test uses SQLAlchemy Core insert/on-conflict:

from sqlalchemy.dialects.postgresql import insert

stmt = insert(my_table).values(id=1, name="x").on_conflict_do_update(
    index_elements=[my_table.c.id],
    set_={"name": stmt.excluded.name},
)

engine = SqlAlchemyConnector.load(name).get_engine()
with engine.begin() as conn:
    conn.execute(stmt)

Docs for reference: - SQLAlchemy 2.0 migration (“Engine.execute removed”): Engine.execute removed - Prefect SQLAlchemy integration methods: Prefect SQLAlchemy integration If you paste the snippet from

tests/io/test_upsert.py

, I’ll give you a minimal diff to fix it.