Pause and resume flow runs

Prefect allows you to halt a flow run with two functions that are similar, but slightly different. When a flow run is paused, code execution is stopped and the process continues to run. When a flow run is suspended, code execution is stopped and so is the process.

​

Pause a flow run

Prefect enables pausing an in-progress flow run for manual approval. Prefect exposes this capability with the pause_flow_run and resume_flow_run functions.

You can call pause_flow_run inside a flow:

from prefect import task, flow, pause_flow_run, resume_flow_run

@task
async def marvin_setup():
    return "a raft of ducks walk into a bar..."


@task
async def marvin_punchline():
    return "it's a wonder none of them ducked!"


@flow
async def inspiring_joke():
    await marvin_setup()
    await pause_flow_run(timeout=600)  # pauses for 10 minutes
    await marvin_punchline()

You can also implement conditional pauses:

from prefect import task, flow, pause_flow_run

@task
def task_one():
    for i in range(3):
        sleep(1)
        print(i)

@flow(log_prints=True)
def my_flow():
    terminal_state = task_one.submit(return_state=True)
    if terminal_state.type == StateType.COMPLETED:
        print("Task one succeeded! Pausing flow run..")
        pause_flow_run(timeout=2) 
    else:
        print("Task one failed. Skipping pause flow run..")

Calling this flow blocks code execution after the first task and waits for resumption to deliver the punchline.

await inspiring_joke()
> "a raft of ducks walk into a bar..."

Resume paused flow runs by clicking Resume in the Prefect UI or calling the resume_flow_run utility through client code.

resume_flow_run(FLOW_RUN_ID)

The paused flow run then finishes.

> "it's a wonder none of them ducked!"

​

Suspend a flow run

Similar to pausing a flow run, Prefect enables suspending an in-progress flow run.

This means you can suspend flow runs to save costs instead of paying for long-running infrastructure. However, when the flow run resumes, the flow code will execute again from the beginning of the flow. We recommend using tasks and task caching to avoid recomputing expensive operations.

Prefect exposes this capability through the suspend_flow_run and resume_flow_run functions, as well as the Prefect UI.

When called inside of a flow, suspend_flow_run immediately suspends execution of the flow run. The flow run is marked as Suspended and will not resume until resume_flow_run is called.

Here is an example of a flow that does not block flow execution while paused. This flow exits after one task, and is rescheduled upon resuming. The stored result of the first task is retrieved instead of rerunning.

from prefect import flow, pause_flow_run, task

@task(persist_result=True)
def foo():
    return 42

@flow(persist_result=True)
def noblock_pausing():
    x = foo.submit()
    pause_flow_run(timeout=30, reschedule=True)
    y = foo.submit()
    z = foo(wait_for=[x])
    alpha = foo(wait_for=[y])
    omega = foo(wait_for=[x, y])

You can suspend flow runs out-of-process by calling suspend_flow_run(flow_run_id=<ID>) or selecting the Suspend button in the Prefect UI or Prefect Cloud.

Resume suspended flow runs can by clicking Resume in the Prefect UI or calling the resume_flow_run utility through client code.

resume_flow_run(FLOW_RUN_ID)

​

Waiting for input when pausing or suspending a flow run

When pausing or suspending a flow run you may want to wait for input from a user. Prefect provides a way to do this by using the pause_flow_run and suspend_flow_run functions. These functions accept a wait_for_input argument, which should be a subclass of prefect.input.RunInput(a Pydantic model). When resuming the flow run, users are required to provide data for this model. After successful validation, the flow run resumes, and the return value of the pause_flow_run or suspend_flow_run is an instance of the model containing the provided data.

Here is an example of a flow that pauses and waits for input from a user:

from prefect import flow, pause_flow_run
from prefect.input import RunInput


class UserNameInput(RunInput):
    name: str


@flow(log_prints=True)
async def greet_user():
    user_input = await pause_flow_run(
        wait_for_input=UserNameInput
    )

    print(f"Hello, {user_input.name}!")

Running this flow creates a flow run. The flow run advances until code execution reaches pause_flow_run, at which point it moves into a Paused state. Execution blocks and waits for resumption.

When resuming the flow run, users are prompted to provide a value for the name field of the UserNameInput model. After successful validation, the flow run resumes, and the return value of the pause_flow_run is an instance of the UserNameInput model containing the provided data.

For more information on receiving input from users when pausing and suspending flow runs, see Send and receive flow run inputs.