Billy McMonagle
04/23/2026, 1:20 AM@sync_compatible decorator and @async_dispatch?Marvin
04/23/2026, 1:20 AMMarvin
04/23/2026, 1:21 AM@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
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
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`:
python
@sync_compatible
async def fetch_data():
return await client.read()
With `@async_dispatch`:
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.Billy McMonagle
04/23/2026, 1:24 AM@async_dispatch - specifically, pause_flow_run. It gives me the following error:
"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?Marvin
04/23/2026, 1:24 AMMarvin
04/23/2026, 1:29 AM@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:
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:
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:
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:
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:
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.