Update for Airflow 2.0 (#34)

* Example for Airflow 2.

* Update chapter 3 for Airflow 2.

* Update chapter 5.

* Updated chapter 8 for Airflow 2.

* Fix black.

* Fix docker + k8s examples for alpha 2.

* Update chapters 11, 16 and 17.

* Update all Docker images to Airflow 2 beta

* Rename user to anonymous admin.

* Rename several dags.

* Update to beta 3.

* Update providers for newest beta.

* Remove backport ref.

* Example for Airflow 2.

* Update chapter 3 for Airflow 2.

* Update chapter 5.

* Updated chapter 8 for Airflow 2.

* Fix black.

* Fix docker + k8s examples for alpha 2.

* Update chapters 11, 16 and 17.

* Update all Docker images to Airflow 2 beta

* Rename user to anonymous admin.

* Rename several dags.

* Update to beta 3.

* Update providers for newest beta.

* Remove backport ref.

* Update chapter 6 to Airflow 2

* Update all chapter 6 code to Airflow 2

* Update all chapter 2 DAGs to Airflow 2

* Update CH4 for Airflow 2

* Update CH12 for Airflow 2

* Updated LDAP example to Airflow 2

* Update CH13 RBAC example to Airflow 2

* Update CH13 secretsbackend example for Airflow 2

* Update CH14 for Airflow 2

* Make links in CH14 readme

* Update for Airflow 2

* Update CH2 for Airflow 2

* Update for Airflow 2

* Add back DAG with name download_rocket_launches to follow book examples

* Add task groups.

* Update chapter 3.

* Add task groups.

* Small updates for chapter 5.

* Small updates to chapter 8.

* Small updates for ch10.

* Remove provide_context.

* Update docker image version.

* Update bash imports.

* Rename remaining old chapters to new numbers.

* Move chapters to top-level.

* Merge chapter 10 folders.

* Fix dummy references.

* Fix readme refs.

* Update CH6 for Airflow 2

* Update CH4 for Airflow 2

* Update CH2 for Airflow 2

* CH12 Airflow 2 updates

* CH13 updates for Airflow 2

* CH12 update for Airfow 2.0

* Fix CH18 docker-compose file

* Update CH2 using new Launch Library API v2

* Restructure ch10, update providers.

* Align CH6 DAG names

* Fix ch10 container names.

* CH7 Update for Airflow 2, fixing bugs along the way

* Update CH9

* Update CH13 for Airflow 2

* DAG serialization settings are obsolete in Airflow 2

* Update CH18 for Airflow 2

* Fix flake8 issues.

Co-authored-by: Julian de Ruiter <[email protected]>
Co-authored-by: Bas Harenslak <[email protected]>

Author: 3 people

Diff for: .flake8

@@ -2,4 +2,4 @@
 max-line-length = 88
 ignore = E203,E501
 per-file-ignores =
-    chapters/chapter18/dags/gcp.py:W503
+    chapter18/dags/gcp.py:W503

Diff for: Makefile

@@ -7,20 +7,19 @@ Code accompanying the Manning book [Data Pipelines with Apache Airflow](https://
 Overall, this repository is structured as follows:
 
 ```
-├── CHANGELOG.md          # Changelog detailing updates to the code.
-├── LICENSE
-├── Makefile              # Helper commands.
-├── README.md             # This readme.
-├── chapters              # Code examples for each of the Chapters.
-├── docker                # Supporting Docker image (containing Airflow).
-└── requirements.txt
+├── chapter01                # Code examples for Chapter 1.
+├── chapter02                # Code examples for Chapter 2.
+├── ...
+├── .pre-commit-config.yaml  # Pre-commit config for the CI.
+├── CHANGELOG.md             # Changelog detailing updates to the code.
+├── LICENSE                  # Code license.
+├── README.md                # This readme.
+└── requirements.txt         # CI dependencies.
 ```
 
-The most interesting parts are probably the *docker* directory and the *chapter* directories under *chapters*.
+The *chapterXX* directories contain the code examples for each specific Chapter.
 
-The *docker* directory contains a custom Airflow image that will be used through out the book.
-
-The *chapter* directories contain the code examples for each specific Chapter. Code for each Chapter is generally structured something like follows:
+Code for each Chapter is generally structured something like follows:
 
 ```
 ├── dags                  # Airflow DAG examples (+ other code).

Diff for: README.md

@@ -2,10 +2,10 @@
 
 import airflow.utils.dates
 from airflow import DAG
-from airflow.operators.dummy_operator import DummyOperator
+from airflow.operators.dummy import DummyOperator
 
 dag = DAG(
-    dag_id="01_umbrella_predictions",
+    dag_id="01_umbrella",
     description="Umbrella example with DummyOperators.",
     start_date=airflow.utils.dates.days_ago(5),
     schedule_interval="@daily",

Diff for: chapters/chapter01/README.md renamed to chapter01/README.md

@@ -9,7 +9,7 @@ x-environment: &airflow_environment
   - AIRFLOW__CORE__STORE_SERIALIZED_DAGS=True
   - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=True
   - AIRFLOW__WEBSERVER__RBAC=False
-x-airflow-image: &airflow_image apache/airflow:1.10.12-python3.8
+x-airflow-image: &airflow_image apache/airflow:2.0.0-python3.8
 # ====================================== /AIRFLOW ENVIRONMENT VARIABLES ======================================
 services:
   postgres:
@@ -25,7 +25,8 @@ services:
     depends_on:
       - postgres
     environment: *airflow_environment
-    command: upgradedb
+    entrypoint: /bin/bash
+    command: -c 'airflow db upgrade && sleep 5 && airflow users create --username admin --password admin --firstname Anonymous --lastname Admin --role Admin --email [email protected]'
   webserver:
     image: *airflow_image
     restart: always

Diff for: chapters/chapter01/dags/01_umbrella_predictions.py renamed to chapter01/dags/01_umbrella.py

@@ -0,0 +1,57 @@
+import json
+import pathlib
+
+import airflow.utils.dates
+import requests
+import requests.exceptions as requests_exceptions
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+dag = DAG(
+    dag_id="download_rocket_launches",
+    description="Download rocket pictures of recently launched rockets.",
+    start_date=airflow.utils.dates.days_ago(14),
+    schedule_interval="@daily",
+)
+
+download_launches = BashOperator(
+    task_id="download_launches",
+    bash_command="curl -o /tmp/launches.json -L 'https://ll.thespacedevs.com/2.0.0/launch/upcoming'",  # noqa: E501
+    dag=dag,
+)
+
+
+def _get_pictures():
+    # Ensure directory exists
+    pathlib.Path("/tmp/images").mkdir(parents=True, exist_ok=True)
+
+    # Download all pictures in launches.json
+    with open("/tmp/launches.json") as f:
+        launches = json.load(f)
+        image_urls = [launch["image"] for launch in launches["results"]]
+        for image_url in image_urls:
+            try:
+                response = requests.get(image_url)
+                image_filename = image_url.split("/")[-1]
+                target_file = f"/tmp/images/{image_filename}"
+                with open(target_file, "wb") as f:
+                    f.write(response.content)
+                print(f"Downloaded {image_url} to {target_file}")
+            except requests_exceptions.MissingSchema:
+                print(f"{image_url} appears to be an invalid URL.")
+            except requests_exceptions.ConnectionError:
+                print(f"Could not connect to {image_url}.")
+
+
+get_pictures = PythonOperator(
+    task_id="get_pictures", python_callable=_get_pictures, dag=dag
+)
+
+notify = BashOperator(
+    task_id="notify",
+    bash_command='echo "There are now $(ls /tmp/images/ | wc -l) images."',
+    dag=dag,
+)
+
+download_launches >> get_pictures >> notify

Diff for: chapters/chapter11/docker-compose.yml renamed to chapter01/docker-compose.yml

@@ -5,8 +5,8 @@
 import requests
 import requests.exceptions as requests_exceptions
 from airflow import DAG
-from airflow.operators.bash_operator import BashOperator
-from airflow.operators.python_operator import PythonOperator
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
 
 dag = DAG(
     dag_id="listing_2_10",
@@ -16,7 +16,7 @@
 
 download_launches = BashOperator(
     task_id="download_launches",
-    bash_command="curl -o /tmp/launches.json 'https://launchlibrary.net/1.4/launch?next=5&mode=verbose'",  # noqa: E501
+    bash_command="curl -o /tmp/launches.json -L 'https://ll.thespacedevs.com/2.0.0/launch/upcoming'",  # noqa: E501
     dag=dag,
 )
 
@@ -28,7 +28,7 @@ def _get_pictures():
     # Download all pictures in launches.json
     with open("/tmp/launches.json") as f:
         launches = json.load(f)
-        image_urls = [launch["rocket"]["imageURL"] for launch in launches["launches"]]
+        image_urls = [launch["image"] for launch in launches["results"]]
         for image_url in image_urls:
             try:
                 response = requests.get(image_url)

Diff for: chapters/chapter02/README.md renamed to chapter02/README.md

@@ -5,18 +5,18 @@
 import requests
 import requests.exceptions as requests_exceptions
 from airflow import DAG
-from airflow.operators.bash_operator import BashOperator
-from airflow.operators.python_operator import PythonOperator
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
 
 dag = DAG(
-    dag_id="listing_2_2",
+    dag_id="listing_2_02",
     start_date=airflow.utils.dates.days_ago(14),
     schedule_interval=None,
 )
 
 download_launches = BashOperator(
     task_id="download_launches",
-    bash_command="curl -o /tmp/launches.json 'https://launchlibrary.net/1.4/launch?next=5&mode=verbose'",  # noqa: E501
+    bash_command="curl -o /tmp/launches.json -L 'https://ll.thespacedevs.com/2.0.0/launch/upcoming'",  # noqa: E501
     dag=dag,
 )
 
@@ -28,7 +28,7 @@ def _get_pictures():
     # Download all pictures in launches.json
     with open("/tmp/launches.json") as f:
         launches = json.load(f)
-        image_urls = [launch["rocket"]["imageURL"] for launch in launches["launches"]]
+        image_urls = [launch["image"] for launch in launches["results"]]
         for image_url in image_urls:
             try:
                 response = requests.get(image_url)

Diff for: chapter02/dags/download_rocket_launches.py

@@ -2,7 +2,7 @@
 from airflow import DAG
 
 dag = DAG(
-    dag_id="listing_2_3",
+    dag_id="listing_2_03",
     start_date=airflow.utils.dates.days_ago(14),
     schedule_interval=None,
 )

Diff for: chapters/chapter02/dags/listing_2_10.py renamed to chapter02/dags/listing_2_10.py

@@ -1,15 +1,15 @@
 import airflow
 from airflow import DAG
-from airflow.operators.bash_operator import BashOperator
+from airflow.operators.bash import BashOperator
 
 dag = DAG(
-    dag_id="listing_2_4",
+    dag_id="listing_2_04",
     start_date=airflow.utils.dates.days_ago(14),
     schedule_interval=None,
 )
 
 download_launches = BashOperator(
     task_id="download_launches",
-    bash_command="curl -o /tmp/launches.json 'https://launchlibrary.net/1.4/launch?next=5&mode=verbose'",  # noqa: E501
+    bash_command="curl -o /tmp/launches.json -L 'https://ll.thespacedevs.com/2.0.0/launch/upcoming'",  # noqa: E501
     dag=dag,
 )

Diff for: chapters/chapter02/dags/listing_2_2.py renamed to chapter02/dags/listing_2_2.py

@@ -5,18 +5,18 @@
 import requests
 import requests.exceptions as requests_exceptions
 from airflow import DAG
-from airflow.operators.bash_operator import BashOperator
-from airflow.operators.python_operator import PythonOperator
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
 
 dag = DAG(
-    dag_id="listing_2_6",
+    dag_id="listing_2_06",
     start_date=airflow.utils.dates.days_ago(14),
     schedule_interval=None,
 )
 
 download_launches = BashOperator(
     task_id="download_launches",
-    bash_command="curl -o /tmp/launches.json 'https://launchlibrary.net/1.4/launch?next=5&mode=verbose'",  # noqa: E501
+    bash_command="curl -o /tmp/launches.json -L 'https://ll.thespacedevs.com/2.0.0/launch/upcoming'",  # noqa: E501
     dag=dag,
 )
 
@@ -28,7 +28,7 @@ def _get_pictures():
     # Download all pictures in launches.json
     with open("/tmp/launches.json") as f:
         launches = json.load(f)
-        image_urls = [launch["rocket"]["imageURL"] for launch in launches["launches"]]
+        image_urls = [launch["image"] for launch in launches["results"]]
         for image_url in image_urls:
             try:
                 response = requests.get(image_url)

Diff for: chapters/chapter02/dags/listing_2_3.py renamed to chapter02/dags/listing_2_3.py

@@ -9,9 +9,8 @@ x-environment: &airflow_environment
   - AIRFLOW__CORE__STORE_DAG_CODE=True
   - AIRFLOW__CORE__STORE_SERIALIZED_DAGS=True
   - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=True
-  - AIRFLOW__WEBSERVER__RBAC=False
 
-x-airflow-image: &airflow_image apache/airflow:1.10.12-python3.8
+x-airflow-image: &airflow_image apache/airflow:2.0.0-python3.8
 # ====================================== /AIRFLOW ENVIRONMENT VARIABLES ======================================
 services:
   postgres:
@@ -22,12 +21,15 @@ services:
       - POSTGRES_DB=airflow
     ports:
       - "5432:5432"
+
   init:
     image: *airflow_image
     depends_on:
       - postgres
     environment: *airflow_environment
-    command: upgradedb
+    entrypoint: /bin/bash
+    command: -c 'airflow db init && airflow users create --username admin --password admin --firstname Anonymous --lastname Admin --role Admin --email [email protected]'
+
   webserver:
     image: *airflow_image
     restart: always
@@ -39,6 +41,7 @@ services:
       - logs:/opt/airflow/logs
     environment: *airflow_environment
     command: webserver
+
   scheduler:
     image: *airflow_image
     restart: always