Improve chapter03

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 '( \

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(

chapter03/dags/02_daily_schedule.py

@@ -1,16 +1,17 @@
-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
 
 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(

chapter03/dags/03_with_end_date.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="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(

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(

chapter03/dags/05_query_with_dates.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="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(

chapter03/dags/06_templated_query.py

@@ -1,17 +1,18 @@
-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="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(

chapter03/dags/07_templated_query_ds.py

@@ -1,18 +1,18 @@
-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
 
 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(

chapter03/dags/08_templated_path.py

@@ -1,17 +1,18 @@
-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="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(

chapter03/dags/09_no_catchup.py

@@ -1,17 +1,18 @@
-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="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,
 )
 

chapter03/dags/10_non_atomic_send.py

@@ -1,17 +1,18 @@
-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="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,
 )
 

chapter03/dags/11_atomic_send.py

@@ -1,17 +1,18 @@
-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="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,
 )
 

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")

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