https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> if I call `run_deployment` in a sync ...
# ask-marvin
t
@Marvin if I call 
run_deployment
in a sync function can I expect my return to be a flowrun object?
m
✅ thought for 131.1 seconds
n
the answer should be yes but unfortunately it only is if you're not calling that sync function from an async context https://github.com/PrefectHQ/prefect/issues/15008 so, to force sync behavior for dual/contextual things like 
run_deployment
you can do 
run_deployment(..., _sync=True)
and it will always behave like a sync function
m
Yes. In Prefect 3.x, 
run_deployment(...)
returns a 
FlowRun
object. - Import path: 
from prefect.deployments import run_deployment
- Return type: 
FlowRun
- You can call it from synchronous code and use the returned object’s attributes (e.g., 
id
, 
state
, 
name
). Example:
Copy code
from prefect.deployments import run_deployment

flow_run = run_deployment("my-flow/my-deployment", parameters={"x": 1})
print(flow_run.id)
# optionally: print(flow_run.state)
API reference: Prefect 3.x deployments API
n
marvin's answer is correct, but if you called 
run_deployment
from a sync function, from another parent async function, the 
@sync_compatible
decorator would have 
run_deployment
return a coro, hence confusion articulated by #15008
❤️ 1
t
@Marvin so if I am calling this sync flow (which is deployed) on the web api (via quick run/custom run), does that make it an async context?
m
✅ thought for 133.9 seconds
Short answer: No. Triggering a deployed flow via the UI (Quick Run / Custom Run) does not “make it an async context.” It just creates a flow run that a worker picks up and executes your flow exactly as you wrote it: - If your flow function is synchronous (
def my_flow(...):
), it runs synchronously. - If your flow is async (
async def my_flow(...):
), it runs in an event loop. If you call 
run_deployment
from a synchronous flow (even one started from the UI), that call is sync-friendly. It returns a 
FlowRun
object, and you can control whether it waits for the child run to finish with the `timeout`/`poll_interval` parameters. It does not require you to be in an async context. Example:
Copy code
from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent():
    # treat the child run as a subflow and wait for completion
    child = run_deployment(
        "my-flow/my-deployment",
        parameters={"x": 1},
        timeout=None,          # wait indefinitely for completion
        poll_interval=5,
        as_subflow=True        # default; shows up nested under the parent
    )
    return child.state
API reference: Prefect 3.x deployments API
upvote 1
t
@Marvin Assuming this is perfectly sync and (hopefully not async): Does this work:
Copy code
if update_db:
        update_flowrun: FlowRun = run_deployment(
            name="update-inference-status",
            parameters={
                "video_path": str(video_path).removeprefix(VAST_PREFIX),
                "status": "running",
                "model_path": str(model_path).removeprefix(VAST_PREFIX),
                "output_path": None,
            },
            as_subflow=True
        )
        while not update_flowrun.state.is_final():
            time.sleep(0.1)
            update_flowrun = FlowRun.load(update_flowrun.id)
m
✅ thought for 379.0 seconds
n
why do you have to poll?
run_deployment
already polls for a terminal state
err i guess you're not polling, but i guess im just saying that 
run_deployment
wont return until its in a terminal state unless you say 
timeout=0
t
ngl claude recommended that, also just want to make sure it’s in a terminal state
ohhhh i see
n
ah yea you should be good w/o the extra while not 
is_final
check
m
Almost! Two important notes: 1) There’s no 
FlowRun.load(...)
in Prefect 3.x, so that line will fail. 2) Prefer the full deployment slug (
"flow-name/deployment-name"
) or the deployment UUID for the 
name
argument, not just the deployment name. If you want to block until the child run finishes, use 
wait_for_flow_run
or set 
timeout
on 
run_deployment
. Here are two working options: Option A: Use wait_for_flow_run (recommended)
Copy code
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run

if update_db:
    fr = run_deployment(
        name="my-flow/update-inference-status",
        parameters={
            "video_path": str(video_path).removeprefix(VAST_PREFIX),
            "status": "running",
            "model_path": str(model_path).removeprefix(VAST_PREFIX),
            "output_path": None,
        },
        as_subflow=True,
    )
    final = wait_for_flow_run(fr.id, timeout=3600, poll_interval=5)
    # final.state is the terminal state
Option B: Poll with the client (sync)
Copy code
import time
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client

if update_db:
    fr = run_deployment(
        name="my-flow/update-inference-status",
        parameters={
            "video_path": str(video_path).removeprefix(VAST_PREFIX),
            "status": "running",
            "model_path": str(model_path).removeprefix(VAST_PREFIX),
            "output_path": None,
        },
        as_subflow=True,
    )

    with get_client(sync_client=True) as client:
        while True:
            fr = client.read_flow_run(fr.id)
            if fr.state.is_final():
                break
            time.sleep(1)
Tip: 
run_deployment
can also handle waiting itself via the 
timeout
parameter if you want it to block; otherwise it returns the 
FlowRun
immediately and you can poll with one of the approaches above. Docs: - run_deployment: Prefect 3.x deployments API - wait_for_flow_run: Prefect 3.x flow_runs API
🙄 1
18 Views
Open in Slack
PreviousNext
Powered by