@Marvin In Prefect 3 what is proper way to reschedule current flow run to another time?

In Prefect 3, if you want to reschedule a flow run to another time, you can use the Prefect client to manage flow runs programmatically. Here’s an example of how you can do this by using a script to reschedule late runs: 1. Use the

Scheduled

state from

prefect.states

. 2. Create an asynchronous function to adjust flow run schedules:

python
from prefect.states import Scheduled
from prefect.client import get_client
from datetime import timedelta, datetime, timezone
from prefect.filters import FlowRunFilter, DeploymentFilter, FlowRunSort

async def reschedule_late_flow_runs(deployment_name: str, delay: timedelta, most_recent_n: int, delete_remaining: bool = True, states: Optional[list[str]] = None) -> list:
    if not states:
        states = ["Late"]
    async with get_client() as client:
        flow_runs = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(
                state=dict(name=dict(any_=states)),
                expected_start_time=dict(
                    before_=datetime.now(timezone.utc)
                ),
            ),
            deployment_filter=DeploymentFilter(
                name={'like_': deployment_name}
            ),
            sort=FlowRunSort.START_TIME_DESC,
            limit=most_recent_n if not delete_remaining else None
        )
        if not flow_runs:
            print(f"No flow runs found in states: {states!r}")
            return []
        rescheduled_flow_runs = []
        for i, run in enumerate(flow_runs):
            await client.delete_flow_run(flow_run_id=run.id)
            if i < most_recent_n:
                # Here you might create a new run or reschedule as needed
                pass

This script finds and optionally deletes flow runs in a certain state (e.g., "Late") and can reschedule those runs as needed. For further details, you can refer to the Prefect 3.0 API interaction docs.

@Marvin But how to schedule runs inside flow code?

In Prefect 3, you typically set up scheduling through deployments rather than in the flow code itself. Prefect provides multiple ways to define schedules for your flows: 1. Prefect UI: You can add, edit, and manage schedules directly in the Prefect UI. 2. CLI: Use the

prefect deployment schedule

command to configure schedules programmatically via the command line. 3. YAML Configuration: Define your schedules in the

prefect.yaml

file associated with your project. This allows you to specify parameters like

cron

,

interval

, or

rrule

when you configure your deployment. 4. Programmatic Deployment: If you're deploying a flow programmatically, you can use the

flow.deploy

or

flow.serve

methods. This is also where you would set parameters for scheduling. Here's an example of how you might define a schedule in the `prefect.yaml`:

yaml
deployments:
  - name: my-scheduled-deployment
    schedule:
      interval: "@daily"  # using cron, interval, or rrule as needed

For more details, you can check out the Prefect 3.0 documentation on adding schedules.

Hi Dzmitry! If you want to schedule child flows from inside a parent flow to run at a later time, you may consider calling run_deployment and using the

scheduled_time

parameter.

The Prefect 3 documentation does not explicitly mention a

scheduled_time

parameter within the

run_deployment

method for programmatically scheduling a flow run in the future. However, scheduling is typically handled by defining a schedule in deployments (like using

cron

,

interval

, or

rrule

schedules) and not directly within the

run_deployment

call. If you want to programmatically schedule a flow run for a future time, you can set up a deployment with the desired schedule using the tools provided by Prefect, such as the CLI, UI, or in the

prefect.yaml

file. You can specify advanced scheduling features like starting a deployment at a future anchor date. If your requirement is to start runs at arbitrary future times not covered by the standard schedule string formats, you may need to manage these through custom scheduling logic or external orchestration to trigger the deployment at specific times. For further details on setting up schedules, refer to Prefect's scheduling documentation.

Eh, he may be missing the sdk docs? 🤔

Hi, I have actually tried it, also with as_subflow=False and timeout=0, but it only works if current flow is running, but I want to complete current run after scheduling new one. I couldn't make it work so.