<@ULVA73B9P> What is the difference between the `@...
# ask-marvin
b
@Marvin What is the difference between the
@sync_compatible
decorator and
@async_dispatch
?
m
thought for 75.5 seconds
In Prefect 3.x, the short version is: -
@sync_compatible
wraps one async implementation and makes it callable from both sync and async contexts. -
@async_dispatch
lets you define separate sync and async implementations and dispatches to the right one automatically.
@async_dispatch
is the newer/preferred pattern.
Example mental model:
@sync_compatible
Copy code
python
@sync_compatible
async def my_fn():
    ...
You write only the async version, and Prefect figures out how to run it in sync code too.
@async_dispatch
Copy code
python
async def amy_fn():
    ...

@async_dispatch(amy_fn)
def my_fn():
    ...
You write both versions explicitly: -
my_fn()
for sync -
amy_fn()
for async Then Prefect dispatches based on context. Why they differ
@sync_compatible
- Best when you already have an async function - Prefect may: - return a coroutine in async contexts - run the coroutine for you in sync contexts - create/manage an event loop if needed - Less explicit - More wrapper magic - Historically useful, but more “legacy” in Prefect 3.x
@async_dispatch
- Best when you want first-class sync and async APIs - Calls the sync implementation in sync contexts - Calls the async implementation in async contexts - More explicit - Better type behavior / clearer intent - This is the direction Prefect has been migrating toward internally Practical guidance Use
@sync_compatible
if: - you have an existing async function and want a quick dual-use wrapper Use
@async_dispatch
if: - you are writing new Prefect-facing APIs - you want separate optimized sync/async code paths - you want the clearest and most maintainable implementation Example With `@sync_compatible`:
Copy code
python
@sync_compatible
async def fetch_data():
    return await client.read()
With `@async_dispatch`:
Copy code
python
async def afetch_data():
    return await client.read()

@async_dispatch(afetch_data)
def fetch_data():
    sync_client = get_client(sync_client=True)
    return sync_client.read()
So the key distinction is: -
@sync_compatible
= one async function, dual-use wrapper
-
@async_dispatch
= two explicit implementations, context-based dispatch
If you want, I can also show the actual source-level behavior of how each one decides whether it’s in a sync vs async context.
b
@Marvin my type checker doesn't like a function decorated with
@async_dispatch
- specifically,
pause_flow_run
. It gives me the following error:
Copy code
"SyncCheckResumeInput" is not awaitable
  "SyncCheckResumeInput" is incompatible with protocol "Awaitable[_T_co@Awaitable]"
    "await" is not present, "SyncCheckResumeInput" is not awaitable
  "SyncCheckResumeInput" is incompatible with protocol "Awaitable[_T_co@Awaitable]"
    "await" is not present
How can I fix this issue?
m
thought for 267.7 seconds
This is a known limitation of using
@async_dispatch
with static type checkers. What’s happening
pause_flow_run
is decorated so that at runtime it dispatches correctly: - in sync code, it uses the sync implementation - in async code, it uses the async implementation But your type checker only sees the declared return type of
pause_flow_run
, which is effectively
T | None
— not an
Awaitable[T | None]
. So this:
Copy code
python
result = await pause_flow_run(wait_for_input=MyInput)
works at runtime in an async flow, but the type checker complains because it thinks you’re trying to
await
a plain
MyInput | None
. Fix In async code, call the explicit async version instead:
Copy code
python
from prefect.flow_runs import apause_flow_run

result = await apause_flow_run(wait_for_input=MyInput)
That is the cleanest and most type-safe fix. Why this works
apause_flow_run
is a normal async function, so type checkers understand that it returns a coroutine and that
await
is valid. Recommended pattern - In async flows/functions: use
await apause_flow_run(...)
- In sync flows/functions: use
pause_flow_run(...)
Example:
Copy code
python
from prefect import flow
from prefect.flow_runs import apause_flow_run
from pydantic import BaseModel

class SyncCheckResumeInput(BaseModel):
    approved: bool

@flow
async def my_flow():
    response = await apause_flow_run(wait_for_input=SyncCheckResumeInput)
    if response and response.approved:
        print("Approved")
If you want to keep using
pause_flow_run
, your fallback is a type-ignore:
Copy code
python
response = await pause_flow_run(wait_for_input=SyncCheckResumeInput)  # type: ignore[misc]
But I would not recommend that if
apause_flow_run
works for your case. Bottom line Use the explicit async function:
Copy code
python
await apause_flow_run(...)
I didn’t find an existing GitHub issue/discussion for this yet, but this behavior matches the current Prefect 3.x implementation of
pause_flow_run
. Related API: -
pause_flow_run(...)
→ sync-dispatched wrapper -
apause_flow_run(...)
→ explicit async function If you want, I can help rewrite your exact flow snippet to use the right sync/async variant cleanly.