Run a DAG on all days of a month except 1 day

I am running DAG to import data from a source. In this case we get a daily incremental data and a full data on a specific day of the month. So we need 2 DAGs, one that can run daily except for the specific day of the month when we get the monthly full data.

For the monthly DAG the schedule is easy to set to run on the specific day. For the daily one we can set it to run daily. However, the DAG will always fail on the day when we get the full data since that day we won't get the incremental data.

So to avoid getting the errors I want to schedule the DAG so that it runs on all days except for that specific day when we get the full data.

Why 2 DAGs?

You can have a single DAG with first task as branch operator that will decide if the run should be incremental or full and proceed accordingly. You can use BranchPythonOperator for that. Simply write a function that extract the day of the month from the logical date of the DAG.

Lets assume on the 1st of each month the DAG will follow full data path and in the rest of the days it will follow incremental path. The DAG code would be:

from datetime import datetime

from airflow.operators.empty import EmptyOperator

from airflow import DAG
from airflow.decorators import task

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 2, 1)
}

with DAG('my_dag', schedule=None, default_args=default_args):

    @task.branch(task_id="branch_task")
    def branch_func(**kwargs):
        logical_date = kwargs["logical_date"]
        day_of_month = logical_date.strftime("%d")
        print(f"logical_date is: {logical_date} day of month is: {day_of_month} ")
        if int(day_of_month) == 1:
            return "full"
        return "incremental"


    full = EmptyOperator(task_id="full")
    inc = EmptyOperator(task_id="incremental")

    branch_func() >> [full, inc]

branch task logs:

If you still prefer the approach of 2 separated DAGs you can use Timetables to customize scheduling as you see fit.