Update airflow requirements for latest cloud composer

Includes update to Pydantic 2.x, which requires a change in the imports of the package.

Author: mjumbewu

airflow/Dockerfile

@@ -1,4 +1,4 @@
-FROM apache/airflow:2.5.3-python3.8
+FROM apache/airflow:2.10.2-python3.11
 
 # install gcloud as root, then switch back to airflow
 USER root
@@ -17,4 +17,4 @@ USER airflow
 
 COPY requirements.txt /tmp/requirements.txt
 
-RUN pip install --no-cache-dir --user -r /tmp/requirements.txt
+RUN pip install --no-cache-dir -r /tmp/requirements.txt

airflow/Dockerfile.composer

@@ -1,4 +1,4 @@
-FROM apache/airflow:2.4.3-python3.8
+FROM apache/airflow:2.10.2-python3.11
 
 # install gcloud as root, then switch back to airflow
 USER root
@@ -15,8 +15,8 @@ RUN gcloud components install gke-gcloud-auth-plugin
 
 USER airflow
 
-COPY requirements-composer-2.6.5-airflow-2.5.3.txt /tmp/requirements-composer-2.6.5-airflow-2.5.3.txt
-RUN pip install --no-cache-dir --user -r /tmp/requirements-composer-2.6.5-airflow-2.5.3.txt
+COPY requirements-composer-2.11.1-airflow-2.10.2.txt /tmp/requirements-composer-2.11.1-airflow-2.10.2.txt
+RUN pip install --no-cache-dir --user -r /tmp/requirements-composer-2.11.1-airflow-2.10.2.txt
 
 COPY requirements.txt /tmp/requirements.txt
 RUN pip install --no-cache-dir --user -r /tmp/requirements.txt

airflow/dags/airtable_loader_v2/generate_gtfs_download_configs.py

@@ -19,7 +19,7 @@
     GTFSDownloadConfigExtract,
     get_fs,
 )
-from pydantic import ValidationError
+from pydantic.v1 import ValidationError
 
 
 def gtfs_datasets_to_extract_configs(

airflow/dags/download_gtfs_schedule_v2/download_schedule_feeds.py

@@ -24,7 +24,7 @@
     get_fs,
     get_latest,
 )
-from pydantic import validator
+from pydantic.v1 import validator
 from requests.exceptions import HTTPError
 
 GTFS_FEED_LIST_ERROR_THRESHOLD = 0.95

airflow/dags/sync_ntd_data_xlsx/scrape_ntd_xlsx_urls.py

@@ -6,7 +6,7 @@
 
 import requests
 from bs4 import BeautifulSoup
-from pydantic import HttpUrl, ValidationError, parse_obj_as
+from pydantic.v1 import HttpUrl, ValidationError, parse_obj_as
 
 from airflow.exceptions import AirflowException
 

airflow/plugins/operators/airtable_to_gcs.py

@@ -7,7 +7,7 @@
 from calitp_data_infra.auth import get_secret_by_name
 from calitp_data_infra.storage import get_fs, make_name_bq_safe
 from pyairtable import Api
-from pydantic import BaseModel
+from pydantic.v1 import BaseModel
 
 from airflow.models import BaseOperator
 

airflow/plugins/operators/blackcat_to_gcs.py

@@ -7,7 +7,7 @@
 import pendulum
 import requests
 from calitp_data_infra.storage import get_fs, make_name_bq_safe
-from pydantic import BaseModel
+from pydantic.v1 import BaseModel
 
 from airflow.models import BaseOperator
 

airflow/plugins/operators/littlepay_raw_sync.py

@@ -16,9 +16,9 @@
     get_fs,
     get_latest_file,
 )
-from pydantic.class_validators import validator
-from pydantic.error_wrappers import ValidationError
-from pydantic.main import BaseModel
+from pydantic.v1.class_validators import validator
+from pydantic.v1.error_wrappers import ValidationError
+from pydantic.v1.main import BaseModel
 from tqdm import tqdm
 from tqdm.contrib.logging import logging_redirect_tqdm
 

airflow/plugins/operators/scrape_ntd_api.py

@@ -11,7 +11,7 @@
     get_fs,
     make_name_bq_safe,
 )
-from pydantic import HttpUrl, parse_obj_as
+from pydantic.v1 import HttpUrl, parse_obj_as
 
 from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator  # type: ignore

airflow/plugins/operators/scrape_ntd_xlsx.py

@@ -12,7 +12,7 @@
     get_fs,
     make_name_bq_safe,
 )
-from pydantic import HttpUrl, parse_obj_as
+from pydantic.v1 import HttpUrl, parse_obj_as
 
 from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator  # type: ignore

airflow/plugins/operators/scrape_state_geoportal.py

@@ -7,7 +7,7 @@
 import pendulum
 import requests
 from calitp_data_infra.storage import PartitionedGCSArtifact, get_fs  # type: ignore
-from pydantic import HttpUrl, parse_obj_as
+from pydantic.v1 import HttpUrl, parse_obj_as
 
 from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator  # type: ignore

airflow/plugins/utils.py

@@ -11,7 +11,7 @@
     PartitionedGCSArtifact,
     fetch_all_in_partition,
 )
-from pydantic import validator
+from pydantic.v1 import validator
 
 SCHEDULE_UNZIPPED_BUCKET = os.environ["CALITP_BUCKET__GTFS_SCHEDULE_UNZIPPED"]
 SCHEDULE_UNZIPPED_BUCKET_HOURLY = os.environ[