Hello, is there any way to stop flows from running...
# ask-community
c
Hello, is there any way to stop flows from running on certain dates?
Am currently doing the following:
Copy code
def is_holiday():

    holidays_list = [...]
    today = pd.to_datetime('today').strftime('%Y-%m-%d')
    
    return True if today in holidays else False
And then wrapping every Prefect task in the below if/else logic:
Copy code
@task
def example_task():
    
    if not is_holiday():
        execute_code()

    else:
        <http://logger.info|logger.info>('Today is holiday.')
Is there anything natively available in Prefect to handle such cases? Am looking at conditional logic, but not quite sure on implementation details
I have a single Prefect script that handles multiple flows / schedules, and want to exclude a certain group of flows from running on certain days
Managed to get a working version below, but not sure why the code in the
with case(is_holiday, True)
block will always execute, regardless of whether the condition is True / False:
Copy code
@task
def is_holiday():
    holidays = [...]
    today = pd.to_datetime('today').strftime('%Y-%m'%d)
    return True if today in holidays else False
Copy code
from prefect import case, Flow

with Flow(...) as flow:
    
    with case(is_holiday, True):
        <http://logger.info|logger.info>('Today is holiday')

    with case(is_holiday, False):
        execute_code()
Prefect flow with multiple schedules - the case function
is_holiday
will run on each trigger of a task run, is this understanding correct ?
Example:
Copy code
from prefect import case, Flow
from prefect.schedules.schedules import Schedule

...

schedule_one = clocks.CronClock("0 15 * * 1-6", parameter_defaults={"report_number": [
    "report_one",
    "report_two",
    "report_three"
]})

schedule_two = clocks.CronClock("0 20 * * 1-6", parameter_defaults={"report_number": [
    "report_four,
    "report_five",
    "report_six"
]})

with Flow(..., schedule=Schedule(clocks=[
    schedule_one,
    schedule_two
]) as flow:

    with case(is_holiday, True):
        <http://logger.info|logger.info>('Today is holiday')
    
    with case(is_holiday, False):
        execute_code()
This means that the case function
is_holiday
would be executed 6 times?
n
Hi @CA Lee - have you seen schedule filters? They should let you filter from your list of dates there.
c
Looks perfect! Haven’t used these yet, will try, thanks
n
😄
c
In case anyone else has the same problem, pass in a list of filters into a Schedule object:
Copy code
from prefect import Flow
from prefect.schedules import filters
from prefect.schedules.schedules import Schedule


def holidays():
    # List of datetime objects
    holidays = [...] 

    # Return a list of filters
    [return filters.on_date(month=date.month, day=date.day) for date in holidays]

with Flow(..., schedule=Schedule(clocks=[
    schedule_one,
    schedule_two], 

    # Pass in list of filters into Schedule
    not_filters=holidays()
) as flow:
    ...
🚀 1
👍 1