time_filters.py 2.17 KB
Newer Older
Aqu's avatar
Aqu committed
1
2
# Jinja filters used in templated fields
# Add it to your dag with `user_defined_filters`.
Aqu's avatar
Aqu committed
3
# Those filters may be accessible everywhere with an Airflow plugin.
Mforns's avatar
Mforns committed
4
5
6
7
8
9
#
# 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.
Aqu's avatar
Aqu committed
10
11
12

filters = {

Mforns's avatar
Mforns committed
13
14
15
    # Presentation
    'to_ds_month': lambda dt: dt.format('YYYY-MM'),
    'to_ds': lambda dt: dt.format('YYYY-MM-DD'),
Mforns's avatar
Mforns committed
16
    'to_ds_nodash': lambda dt: dt.format('YYYYMMDD'),
17
    'to_ds_hour_nodash': lambda dt: dt.format('YYYYMMDDHH'),
18
    'to_time_nodash': lambda dt: dt.format('HHmmss'),
Aqu's avatar
Aqu committed
19
20

    # Month
Mforns's avatar
Mforns committed
21
22
23
24
25
26
    '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'),
Aqu's avatar
Aqu committed
27
28

    # Week
Mforns's avatar
Mforns committed
29
30
31
32
33
34
    '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'),
35
36

    # Day
Mforns's avatar
Mforns committed
37
38
39
40
41
42
    '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'),
43

Mforns's avatar
Mforns committed
44
45
46
    # Delta
    'add_days': lambda dt, n: dt.add(days=n),
    'add_hours': lambda dt, n: dt.add(hours=n),
Aqu's avatar
Aqu committed
47
}