Make scheduled flows not overlap in time

[igonro]

I've read this in the docs:

Schedules are independent; if schedules overlap, multiple flow runs will be created

But, there is no way to make runs of the same type to not overlap, and cancel the scheduled run if the previous one didn't finish?
If this cannot be done with Prefect, what approach should I follow?

Thanks for your time!

[anna-geller]

Here is an example state handler that allows you to skip (or cancel) a flow run if there are already some flow runs in progress. This allows non-overlapping flow runs, i.e. only one flow run of a specific flow at a time.

import prefect
from prefect import task, Flow
from prefect.client import Client
from prefect.engine.state import Skipped
import time


@task(log_stdout=True)
def hello_world():
    print("Sleeping...")
    time.sleep(360)


def skip_if_running_handler(obj, old_state, new_state):
    if new_state.is_running():
        client = Client()
        query = """
            query($flow_id: uuid) {
              flow_run(
                where: {_and: [{flow_id: {_eq: $flow_id}},
                {state: {_eq: "Running"}}]}
                limit: 1
                offset: 1
              ) {
                name
                state
                start_time
              }
            }
        """
        response = client.graphql(
            query=query, variables=dict(flow_id=prefect.context.flow_id)
        )
        active_flow_runs = response["data"]["flow_run"]
        if active_flow_runs:
            logger = prefect.context.get("logger")
            message = "Skipping this flow run since there are already some flow runs in progress"
            logger.info(message)
            return Skipped(message) # or returned Cancelled state if you prefer this state in this use case
    return new_state


with Flow("skip_if_running", state_handlers=[skip_if_running_handler]) as flow:
    hello_task = hello_world()

[igonro]

Hi @anna-geller. Thank you so much for this detailed response! I think this will do it.

[alexander-belikov]

Hi @igonro , did you manage to make use of the snippet above?

For some reason in my case it does not prevent starting the same flow.

[igonro]

Hi @alexander-belikov

Yes, it is working for us, we are using this code:

def skip_if_running_handler(obj, old_state, new_state):
    if new_state.is_running():
        client = Client()
        query = (
            "query($flow_id: uuid){flow_run(where:{_and:[{flow_id:{_eq:$flow_id}}",
            '{state:{_eq:"Running"}}]}){name state start_time}}',
        )
        response = client.graphql(query=query, variables=dict(flow_id=prefect.context.flow_id))
        active_flow_runs = response["data"]["flow_run"]
        if active_flow_runs:
            return Skipped("Skipping this flow run since there are already some flow runs in progress")
    return new_state

Check if the query works for you, we removed some params (offset and limit), but I don't remember exactly why.

[alexander-belikov]

Thank you so much for such a prompt response!
So far I have not been successful. Maybe it has to do with how I register the flows.
I'm using the CLI as

prefect register --project pname --path script.py --name flowname

Perhaps I'm missing something in the setup of the prefect service?

[anna-geller]

hard to say without knowing more details

this command works already without flow name: prefect register --project pname --path script.py

[alexander-belikov]

Thank @igonro and @anna-geller!
I'm not sure why it didn't work initially. It works as a charm now and it is a very useful functionality.

[t1g0rz]

How can the same task be resolved in Prefect v2?