time_filters.py

# Jinja filters used in templated fields
# Add it to your dag with `user_defined_filters`.
# Those filters may be accessible everywhere with an Airflow plugin.
#
# NOTE: The end_of() modifiers return the datetime at the last microsecond
# of the corresponding period. For instance. datetime(2022, 1, 1).end_of('day')
# will return 2022-01-01T23:59:59.999999. Thus, when using end_of() to i.e.
# specify the range of partitions to wait for, Airflow might leave out the
# last partition of the period. Consider using start_of() filters instead.

filters = {

    # Presentation
    'to_ds_month': lambda dt: dt.format('YYYY-MM'),
    'to_ds': lambda dt: dt.format('YYYY-MM-DD'),
    'to_ds_nodash': lambda dt: dt.format('YYYYMMDD'),
    'to_ds_hour_nodash': lambda dt: dt.format('YYYYMMDDHH'),
    'to_time_nodash': lambda dt: dt.format('HHmmss'),

    # Month
    'start_of_current_month': lambda dt: dt.start_of('month'),
    'end_of_current_month': lambda dt: dt.end_of('month'),
    'start_of_previous_month': lambda dt: dt.subtract(months=1).start_of('month'),
    'end_of_previous_month': lambda dt: dt.subtract(months=1).end_of('month'),
    'start_of_next_month': lambda dt: dt.add(months=1).start_of('month'),
    'end_of_next_month': lambda dt: dt.add(months=1).end_of('month'),

    # Week
    'start_of_current_week': lambda dt: dt.start_of('week'),
    'end_of_current_week': lambda dt: dt.end_of('week'),
    'start_of_previous_week': lambda dt: dt.subtract(weeks=1).start_of('week'),
    'end_of_previous_week': lambda dt: dt.subtract(weeks=1).end_of('week'),
    'start_of_next_week': lambda dt: dt.add(weeks=1).start_of('week'),
    'end_of_next_week': lambda dt: dt.add(weeks=1).end_of('week'),

    # Day
    'start_of_current_day': lambda dt: dt.start_of('day'),
    'end_of_current_day': lambda dt: dt.end_of('day'),
    'start_of_previous_day': lambda dt: dt.subtract(days=1).start_of('day'),
    'end_of_previous_day': lambda dt: dt.subtract(days=1).end_of('day'),
    'start_of_next_day': lambda dt: dt.add(days=1).start_of('day'),
    'end_of_next_day': lambda dt: dt.add(days=1).end_of('day'),

    # Delta
    'add_days': lambda dt, n: dt.add(days=n),
    'add_hours': lambda dt, n: dt.add(hours=n),
}