Has anyone figured out how to overwrite parameters in prefect 2 for a flow on a per schedule basis, the way you could in prefect 1? A lot of overhead in Prefect 2 right now if you just want to send a simple variable change for a flow depending on the time the flow runs. Currently, the only solution I've found is to have 2 separate deployments registered with different schedules and variables.

Hi Seth, can you tell me more about your use case? What are the two different schedules? i.e. what is the condition for which the param would change.

Absolutely. We have to run a flow that imports various items from different services. About 3/4 of the code is shared, but some services only upload their data once a day, while others are multiple times a day. We'd like this flow to run 3-4 times a day, with some variables set with different values to target those services.

At the moment, you can only have one schedule for each deployment. While you can override default params of a deployment for single flow run, alternative solutions to just creating a second deployment would be to use

create_flow_run_from_deployment()

and pass in the custom params but I don’t think that would be more optimal. You could also create a block to do some check to see if the service has updated their data and that way the params could remain unchanged while the flow would ignore the services that hadn’t been updated yet. Those are some solutions off the top of my head anyway.

Hi Seth, I agree with Serina that prefect get_client() is a valid option for setting up a variety of flow runs that run at different schedules with different parameters. (Assuming the flow code is similar across them all) Check out this task, I’m not sure if something like this might work for you:

from prefect import flow, get_run_logger, task
from prefect.client import get_client

@task
async def add_new_scheduled_run(depl_id, delta_hours=6, parameter_dict):
    """
    This task adds a scheduled flow run x hours from now with a specific parameter dict
    """
    # Get the time x hours from now.
    scheduled_time = now.add(hours=delta_hours)

    # Use Prefect get_client() to schedule a new flow run x hours from now
    async with get_client() as client:
        response = await client.create_flow_run_from_deployment(
            deployment_id=depl_id, parameters=parameter_dict state=Scheduled(scheduled_time=scheduled_time)
        )
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"INFO get client response: {response}")
    <http://logger.info|logger.info>(f"INFO Scheduled a flow run for {scheduled_time}!")

Essentially, in using

create_flow_run_from_deployment()

you can add a new schedule with specified parameters to any deployment.

So in this example, I could use that in my flow-of-flows, allowing it to run once in a day and schedule each run with the required parameters.

Right, check out this example for one way you could set up a flow of flow pattern. https://discourse.prefect.io/t/how-to-create-a-flow-run-from-deployment-orchestrator-pattern/803