Upgrade airflow version

.github/workflows/foundry.okd.on-push.application.yml

@@ -191,6 +191,8 @@ jobs:
             --set-string secrets.bcwatFlowworks.password="${{ secrets.FOUNDRY_BCWAT_FLOWWORKS_PASSWORD }}" \
             --set-string secrets.bcwatFlowworks.username="${{ secrets.FOUNDRY_BCWAT_FLOWWORKS_USERNAME }}" \
             --set-string secrets.fernet.key="${{ secrets.FOUNDRY_AIRFLOW_FERNET_KEY }}" \
+            --set-string secrets.api.secretKey="${{ secrets.FOUNDRY_AIRFLOW_API_SECRET_KEY_}}" \
+            --set-string secrets.api.jwtKey="${{ secrets.FOUNDRY_AIRFLOW_API_JWT_KEY_}}" \
             --set-string secrets.sendgridApiKey="${{ secrets.FOUNDRY_SENDGRID_API_KEY }}" \
             --set migrations.image=${{ secrets.ACR_REGISTRY }}/bcwat/flyway:${{ needs.bump_version.outputs.version }} \
             --set-string webserver.podAnnotations."rollout-timestamp"="$(date +%s)" \

airflow/Dockerfile

@@ -1,4 +1,4 @@
-FROM apache/airflow:2.10.5-python3.12
+FROM apache/airflow:3.1.7-python3.12
 
 USER root
 
@@ -14,7 +14,15 @@ ENV PYTHONPATH = "/opt/airflow/etl_pipelines:${PYTHONPATH}"
 
 COPY requirements.txt .
 
+# Switch to airflow user for pip installations
 USER airflow
 
+# PIN AIRFLOW VERSION when installing packages
+RUN pip install --no-cache-dir "apache-airflow==${AIRFLOW_VERSION}" \
+    --constraint ${HOME}/constraints.txt \
+    apache-airflow-providers-cncf-kubernetes \
+    apache-airflow-providers-sendgrid
+
+RUN pip install --no-cache-dir apache-airflow-providers-fab==3.1.2
+
 RUN pip install --no-cache-dir -r requirements.txt
-RUN pip install apache-airflow-providers-cncf-kubernetes apache-airflow-providers-sendgrid

airflow/Dockerfile_OKD

@@ -1,4 +1,4 @@
-FROM apache/airflow:2.10.5-python3.12
+FROM apache/airflow:3.1.7-python3.12
 
 USER root
 
@@ -10,12 +10,18 @@ COPY etl_pipelines/ /opt/airflow/etl_pipelines/
 
 RUN chown -R airflow: /opt/airflow/dags /opt/airflow/etl_pipelines /opt/airflow/shared
 
-
 ENV PYTHONPATH = "/opt/airflow/etl_pipelines:${PYTHONPATH}"
 
 COPY requirements.txt .
 
 USER airflow
 
+# PIN AIRFLOW VERSION when installing packages
+RUN pip install --no-cache-dir "apache-airflow==${AIRFLOW_VERSION}" \
+    --constraint ${HOME}/constraints.txt \
+    apache-airflow-providers-cncf-kubernetes \
+    apache-airflow-providers-sendgrid
+
+RUN pip install --no-cache-dir apache-airflow-providers-fab==3.1.2
+
 RUN pip install --no-cache-dir -r requirements.txt
-RUN pip install apache-airflow-providers-cncf-kubernetes apache-airflow-providers-sendgrid

airflow/README.md

@@ -29,7 +29,7 @@ In production, we are using the `KubernetesExecutor`. This does not impact runni
 ```bash
 import os
 from datetime import datetime
-from airflow.decorators import dag, task
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -41,7 +41,7 @@ ENVIRONMENT = os.getenv('ENVIRONMENT', 'no-env-found')
 # Does not prevent running locally
 # pod_template_file handles worker pod config
 @dag(
-    schedule_interval="@daily",
+    schedule="@daily",
     start_date=datetime(2024, 1, 1),
     catchup=False,
     tags=["example"],

airflow/dags/asp_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="asp_dag",
-    schedule_interval="5 8 * * *",
-    start_date=pendulum.datetime(2025, 5, 7, tz="UTC"),
+    schedule="5 8 * * *",
+    start_date=datetime(2025, 5, 7),
     catchup=False,
     tags=["water","climate", "station_observations", "daily"],
     default_args=default_args

airflow/dags/ec_xml_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="ec_xml_dag",
-    schedule_interval="0 8 * * *",
-    start_date=pendulum.datetime(2025, 5, 7, tz="UTC"),
+    schedule="0 8 * * *",
+    start_date=datetime(2025, 5, 7),
     catchup=False,
     tags=["climate", "station_observations", "daily"],
     default_args=default_args

airflow/dags/env_aqn_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="env_aqn_dag",
-    schedule_interval="0 8 * * *",
-    start_date=pendulum.datetime(2025, 5, 15, tz="UTC"),
+    schedule="0 8 * * *",
+    start_date=datetime(2025, 5, 15),
     catchup=False,
     tags=["climate", "station_observations", "daily"],
     default_args=default_args

airflow/dags/env_hydro_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="env_hydro_dag",
-    schedule_interval="10 8 * * *",
-    start_date=pendulum.datetime(2025, 4, 17, tz="UTC"),
+    schedule="10 8 * * *",
+    start_date=datetime(2025, 4, 17),
     catchup=False,
     tags=["water", "station_observations", "daily"],
     default_args=default_args

airflow/dags/flnro_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="flnro_wmb_dag",
-    schedule_interval="0 8 * * *",
-    start_date=pendulum.datetime(2025, 5, 15, tz="UTC"),
+    schedule="0 8 * * *",
+    start_date=datetime(2025, 5, 15),
     catchup=False,
     tags=["climate", "station_observations", "daily"],
     default_args=default_args

airflow/dags/flow_works_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="flowworks_dag",
-    schedule_interval="0 8 * * *",
-    start_date=pendulum.datetime(2025, 4, 17, tz="UTC"),
+    schedule="0 8 * * *",
+    start_date=datetime(2025, 4, 17),
     catchup=False,
     tags=["water","climate", "station_observations", "daily"],
     default_args=default_args

airflow/dags/gw_moe_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="gw_moe_dag",
-    schedule_interval="15 8 * * *",
-    start_date=pendulum.datetime(2025, 4, 17, tz="UTC"),
+    schedule="15 8 * * *",
+    start_date=datetime(2025, 4, 17),
     catchup=False,
     tags=["groundwater", "station_observations", "daily"],
     default_args=default_args

airflow/dags/msp_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="msp_dag",
-    schedule_interval="0 8 * * *",
-    start_date=pendulum.datetime(2025, 5, 7, tz="UTC"),
+    schedule="0 8 * * *",
+    start_date=datetime(2025, 5, 7),
     catchup=False,
     tags=["water","climate", "station_observations", "daily"],
     default_args=default_args

airflow/dags/quarterly_climate_ec_update_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -11,8 +11,8 @@
 @dag(
     dag_id="quarterly_ec_update_dag",
     # Cron for At 08:30 UTC (00:30 PST) on day-of-month 1 in every 3rd month.
-    schedule_interval="30 8 1 */3 *",
-    start_date=pendulum.datetime(2025, 6, 13, tz="UTC"),
+    schedule="30 8 1 */3 *",
+    start_date=datetime(2025, 6, 13),
     catchup=False,
     tags=["climate", "quarterly"],
     default_args=default_args

airflow/dags/quarterly_ems_water_quality_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -11,8 +11,8 @@
 @dag(
     dag_id="quarterly_ems_water_quality_dag",
     # Cron for At 08:30 UTC (00:30 PST) on day-of-month 2 in every 3rd month.
-    schedule_interval="30 8 2 */3 *",
-    start_date=pendulum.datetime(2025, 7, 3, tz="UTC"),
+    schedule="30 8 2 */3 *",
+    start_date=datetime(2025, 7, 3),
     catchup=False,
     tags=["waterquality", "quarterly"],
     default_args=default_args

airflow/dags/quarterly_gw_moe_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -11,8 +11,8 @@
 @dag(
     dag_id="quarterly_moe_gw_update",
     # Cron for At 09:00 UTC (01:00 PST) on day-of-month 1 in every 3rd month.
-    schedule_interval="0 9 1 */3 *",
-    start_date=pendulum.datetime(2025, 6, 13, tz="UTC"),
+    schedule="0 9 1 */3 *",
+    start_date=datetime(2025, 6, 13),
     catchup=False,
     tags=["groundwater", "quarterly"],
     default_args=default_args

airflow/dags/quarterly_hydat_import_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -13,8 +13,8 @@
     # Cron for At 01:30 on day-of-month 1 and 15 of each month.
     # This was done instead of checking every quarter because there is not consistent schedule for Hydat. If there is not a new version of
     # Hydat available. It will not scrape it.
-    schedule_interval="30 9 1,15 * *",
-    start_date=pendulum.datetime(2025, 6, 13, tz="UTC"),
+    schedule="30 9 1,15 * *",
+    start_date=datetime(2025, 6, 13),
     catchup=False,
     tags=["water", "quarterly", "hydat"],
     default_args=default_args

airflow/dags/quarterly_moe_hydrometric_historic_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -11,8 +11,8 @@
 @dag(
     dag_id="quarterly_moe_hydrometric_historic_update_dag",
     # Cron for At 10:00 UTC (02:00 PST) on day-of-month 1 in every 3rd month.
-    schedule_interval="0 10 1 */3 *",
-    start_date=pendulum.datetime(2025, 6, 13, tz="UTC"),
+    schedule="0 10 1 */3 *",
+    start_date=datetime(2025, 6, 13),
     catchup=False,
     tags=["water", "quarterly"],
     default_args=default_args

airflow/dags/quarterly_water_quality_eccc_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -11,8 +11,8 @@
 @dag(
     dag_id="quarterly_water_quality_eccc_dag",
     # Cron for At 10:15 UTC (02:15 PST) on day-of-month 1 in every 3rd month.
-    schedule_interval="15 10 1 */3 *",
-    start_date=pendulum.datetime(2025, 6, 13, tz="UTC"),
+    schedule="15 10 1 */3 *",
+    start_date=datetime(2025, 6, 13),
     catchup=False,
     tags=["waterquality", "quarterly"],
     default_args=default_args

airflow/dags/update_station_year_var_status_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="update_station_year_var_status_dag",
-    schedule_interval="30 13 * * *",
-    start_date=pendulum.datetime(2025, 7, 15, tz="UTC"),
+    schedule="30 13 * * *",
+    start_date=datetime(2025, 7, 15),
     catchup=False,
     tags=["utility", "daily"],
     default_args=default_args

airflow/dags/water_approval_dag.py

@@ -1,6 +1,6 @@
 import os
-import pendulum
-from airflow.decorators import dag, task
+from datetime import datetime
+from airflow.sdk import dag, task
 from shared.constants import default_args
 from shared.functions import generate_executor_config_template
 from dotenv import load_dotenv, find_dotenv
@@ -10,8 +10,8 @@
 
 @dag(
     dag_id="wls_water_approval_dag",
-    schedule_interval="0 6 * * *",
-    start_date=pendulum.datetime(2025, 5, 29, tz="UTC"),
+    schedule="0 6 * * *",
+    start_date=datetime(2025, 5, 29),
     catchup=False,
     tags=["licence", "databc", "daily"],
     default_args=default_args