diff --git a/chapter02/scripts/listing_2_10.sh b/chapter02/scripts/listing_2_10.sh index 8cfe432e..d390097c 100755 --- a/chapter02/scripts/listing_2_10.sh +++ b/chapter02/scripts/listing_2_10.sh @@ -12,7 +12,7 @@ docker run \ -ti \ -p 8080:8080 \ -v ${SCRIPT_DIR}/../dags/download_rocket_launches.py:/opt/airflow/dags/download_rocket_launches.py \ ---name airflow +--name airflow \ --entrypoint=/bin/bash \ apache/airflow:2.0.0-python3.8 \ -c '( \ diff --git a/chapter03/dags/01_unscheduled.py b/chapter03/dags/01_unscheduled.py index fc7b7919..77f32b1f 100644 --- a/chapter03/dags/01_unscheduled.py +++ b/chapter03/dags/01_unscheduled.py @@ -1,13 +1,13 @@ -from datetime import datetime from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator dag = DAG( - dag_id="01_unscheduled", start_date=datetime(2019, 1, 1), schedule_interval=None + dag_id="01_unscheduled", start_date=airflow.utils.dates.days_ago(14), schedule_interval=None ) fetch_events = BashOperator( diff --git a/chapter03/dags/02_daily_schedule.py b/chapter03/dags/02_daily_schedule.py index 764d08fd..5d12f1c7 100644 --- a/chapter03/dags/02_daily_schedule.py +++ b/chapter03/dags/02_daily_schedule.py @@ -1,7 +1,8 @@ -from datetime import datetime +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator @@ -9,8 +10,8 @@ dag = DAG( dag_id="02_daily_schedule", schedule_interval="@daily", - start_date=datetime(2019, 1, 1), - end_date=datetime(2019, 1, 5), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), ) fetch_events = BashOperator( diff --git a/chapter03/dags/03_with_end_date.py b/chapter03/dags/03_with_end_date.py index 3bc628ef..af283d8e 100644 --- a/chapter03/dags/03_with_end_date.py +++ b/chapter03/dags/03_with_end_date.py @@ -1,7 +1,8 @@ -import datetime as dt +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator @@ -9,8 +10,8 @@ dag = DAG( dag_id="03_with_end_date", schedule_interval="@daily", - start_date=dt.datetime(year=2019, month=1, day=1), - end_date=dt.datetime(year=2019, month=1, day=5), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), ) fetch_events = BashOperator( diff --git a/chapter03/dags/04_time_delta.py b/chapter03/dags/04_time_delta.py index c4287771..d25ef20a 100644 --- a/chapter03/dags/04_time_delta.py +++ b/chapter03/dags/04_time_delta.py @@ -1,16 +1,17 @@ -import datetime as dt +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator dag = DAG( dag_id="04_time_delta", - schedule_interval=dt.timedelta(days=3), - start_date=dt.datetime(year=2019, month=1, day=1), - end_date=dt.datetime(year=2019, month=1, day=5), + schedule_interval=timedelta(days=3), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), ) fetch_events = BashOperator( diff --git a/chapter03/dags/05_query_with_dates.py b/chapter03/dags/05_query_with_dates.py index 39ff4022..c84b69cf 100644 --- a/chapter03/dags/05_query_with_dates.py +++ b/chapter03/dags/05_query_with_dates.py @@ -1,7 +1,8 @@ -import datetime as dt +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator @@ -9,8 +10,8 @@ dag = DAG( dag_id="05_query_with_dates", schedule_interval="@daily", - start_date=dt.datetime(year=2019, month=1, day=1), - end_date=dt.datetime(year=2019, month=1, day=5), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), ) fetch_events = BashOperator( diff --git a/chapter03/dags/06_templated_query.py b/chapter03/dags/06_templated_query.py index 6aef046f..9da20c9c 100644 --- a/chapter03/dags/06_templated_query.py +++ b/chapter03/dags/06_templated_query.py @@ -1,8 +1,9 @@ -import datetime as dt +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator @@ -10,8 +11,8 @@ dag = DAG( dag_id="06_templated_query", schedule_interval="@daily", - start_date=dt.datetime(year=2019, month=1, day=1), - end_date=dt.datetime(year=2019, month=1, day=5), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), ) fetch_events = BashOperator( diff --git a/chapter03/dags/07_templated_query_ds.py b/chapter03/dags/07_templated_query_ds.py index fc4fce45..7105afc6 100644 --- a/chapter03/dags/07_templated_query_ds.py +++ b/chapter03/dags/07_templated_query_ds.py @@ -1,9 +1,9 @@ -import datetime as dt -from datetime import timedelta +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator @@ -11,8 +11,8 @@ dag = DAG( dag_id="07_templated_query_ds", schedule_interval=timedelta(days=3), - start_date=dt.datetime(year=2019, month=1, day=1), - end_date=dt.datetime(year=2019, month=1, day=5), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), ) fetch_events = BashOperator( diff --git a/chapter03/dags/08_templated_path.py b/chapter03/dags/08_templated_path.py index 9b382f48..7e05fbb4 100644 --- a/chapter03/dags/08_templated_path.py +++ b/chapter03/dags/08_templated_path.py @@ -1,8 +1,9 @@ -import datetime as dt +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator @@ -10,8 +11,8 @@ dag = DAG( dag_id="08_templated_path", schedule_interval="@daily", - start_date=dt.datetime(year=2019, month=1, day=1), - end_date=dt.datetime(year=2019, month=1, day=5), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), ) fetch_events = BashOperator( diff --git a/chapter03/dags/09_no_catchup.py b/chapter03/dags/09_no_catchup.py index d86c8d92..f45586de 100644 --- a/chapter03/dags/09_no_catchup.py +++ b/chapter03/dags/09_no_catchup.py @@ -1,8 +1,9 @@ -import datetime as dt +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator @@ -10,8 +11,8 @@ dag = DAG( dag_id="09_no_catchup", schedule_interval="@daily", - start_date=dt.datetime(year=2019, month=1, day=1), - end_date=dt.datetime(year=2019, month=1, day=5), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), catchup=False, ) diff --git a/chapter03/dags/10_non_atomic_send.py b/chapter03/dags/10_non_atomic_send.py index 0ffb08bc..9505ab45 100644 --- a/chapter03/dags/10_non_atomic_send.py +++ b/chapter03/dags/10_non_atomic_send.py @@ -1,8 +1,9 @@ -import datetime as dt +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator @@ -10,8 +11,8 @@ dag = DAG( dag_id="10_non_atomic_send", schedule_interval="@daily", - start_date=dt.datetime(year=2019, month=1, day=1), - end_date=dt.datetime(year=2019, month=1, day=5), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), catchup=True, ) diff --git a/chapter03/dags/11_atomic_send.py b/chapter03/dags/11_atomic_send.py index ce1f16a4..e77ffcb7 100644 --- a/chapter03/dags/11_atomic_send.py +++ b/chapter03/dags/11_atomic_send.py @@ -1,8 +1,9 @@ -import datetime as dt +from datetime import datetime, timedelta from pathlib import Path import pandas as pd +import airflow from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator @@ -10,8 +11,8 @@ dag = DAG( dag_id="11_atomic_send", schedule_interval="@daily", - start_date=dt.datetime(year=2019, month=1, day=1), - end_date=dt.datetime(year=2019, month=1, day=5), + start_date=airflow.utils.dates.days_ago(14), + end_date=datetime.now() + timedelta(days=14), catchup=True, ) diff --git a/chapter03/docker/events-api/app.py b/chapter03/docker/events-api/app.py index aad766c4..e6ba7eb3 100644 --- a/chapter03/docker/events-api/app.py +++ b/chapter03/docker/events-api/app.py @@ -48,7 +48,7 @@ def _generate_events_for_day(date): app = Flask(__name__) -app.config["events"] = _generate_events(end_date=date(year=2019, month=1, day=5)) +app.config["events"] = _generate_events(end_date=date.today()) @app.route("/events") diff --git a/chapter03/docker/events-api/requirements.txt b/chapter03/docker/events-api/requirements.txt index eb024fed..e82b29bd 100644 --- a/chapter03/docker/events-api/requirements.txt +++ b/chapter03/docker/events-api/requirements.txt @@ -1,4 +1,3 @@ -click==7.1.2 faker==4.14.0 -flask==1.1.2 +flask==2.2.2 pandas==1.1.3