CODEWITHSUNDEEP
pythonjinja2airflow
KoKu
KoKu

Reputation: 43

How to get execution_date as epoch time?

In regular python code I can do:

import time
int(time.time())

This gives me the time as epoch. I want to be able to do this with airflow macro: execution_date

This is what I tried:

"{{  strptime(execution_date.strftime('%Y-%m-%d %H:%M:%S'), '%d.%m.%Y %H:%M:%S') }}"

But this gives:

jinja2.exceptions.UndefinedError: 'strptime' is undefined

I'm running Airflow 1.9 & Python 2.7

Upvotes: 4

Views: 7736

Answers (2)

Bas Harenslak
Bas Harenslak

Reputation: 3064

Since Airflow 1.10, Airflow uses Pendulum for datetimes, which has attributes timestamp(), int_timestamp and float_timestamp which return the epoch.

So, you could do:

{{ execution_date.int_timestamp }}

Docs: https://pendulum.eustace.io/docs/#attributes-and-properties

Other options are:

{{ execution_date.strftime('%s') }} # Also Python 2
{{ execution_date.timestamp() }}    # Python >=3.3

Upvotes: 5

Steve McCartney
Steve McCartney

Reputation: 301

There are several approaches to solving this, but first, you should simplify how you are accessing the timestamp.

As you are using Airflow, where execution_date is a pendulum.Pendulum object, you have access to the execution_date.int_timestamp property.

In a more general situation assuming you are using Python 3.3+ there is a simpler way to get the epoch timestamp using the timestamp() method on a datetime object (which is also available on a Pendulum object).

execution_date.int_timestamp
Output => 1555477775

Python 3.3+:
int(execution_date.timestamp())
Output => 1555477775

Python 2.7+:
int(v.strftime('%s'))
Output => 1555477775

1. Direct formatting within the template

This method uses the timestamp() method of the datetime object to generate a float which then run through the built-in int filter of the Jinja2 template.

For Pendulum:
{{ execution_date.int_timestamp }}

For general datetime objects:
{{ execution_date.timestamp()|int }}

2. Create a custom Jinja2 Filter

If you are using this transformation in multiple places throughout your templates, it is probably worth creating a filter to centralise this. With Airflow, you can register a custom filter on the DAG object.

import time

def epoch_int(v):
    return int(v.strftime('%s'))

dag = DAG(
  ...
  user_defined_filters={
    "epoch_int": epoch_int,
  }
  ...
)

and then use it in your template.

{{ execution_date|epoch_int }}

Upvotes: 3

Related Questions