dags to airflow 3

Author: james-otten

dags/example.py

@@ -3,16 +3,15 @@
 import time
 from builtins import range
 from pprint import pprint
+import datetime
 
-from airflow.utils.dates import days_ago
-
-from airflow.models import DAG
-from airflow.operators.python_operator import PythonOperator
-from airflow.operators.bash_operator import BashOperator
+from airflow.sdk import DAG
+from airflow.providers.standard.operators.python import PythonOperator
+from airflow.providers.standard.operators.bash import BashOperator
 
 args = {
     'owner': 'Airflow',
-    'start_date': days_ago(2),
+    'start_date': datetime.datetime(2021, 1, 1),
 }
 
 dag = DAG(

dags/nycmesh_generate_omni_cert.py

@@ -1,6 +1,7 @@
-from airflow.models import DAG
-from airflow.decorators import task
-from airflow.utils.dates import days_ago
+import datetime
+
+from airflow.sdk import dag, task
+
 
 def generate_certbot_tsig_cert(fqdn_string, dns_server, tsig_key_name, tsig_key, full_chain_path, priv_key_path):
     tsig_ini_file_path = "/tsig.ini"
@@ -62,15 +63,16 @@ def deploy_to_omni(ip, password, cert_path, priv_key_path):
 
 args = {
     'owner': 'Airflow',
-    'start_date': days_ago(2),
+    'start_date': datetime.datetime(2021, 1, 1),
 }
 
-with DAG(
+@dag(
     dag_id="omni_nn_certv1",
     default_args=args,
     schedule_interval=None,
     tags=["james", "nn", "tsig", "cert", "device"]
-) as dag:
+)
+def omni_cert_dag():
     @task.virtualenv(
         task_id="certbot_omni_nn_certv1", requirements=["certbot", "paramiko==3.5.0", "scp==0.15.0"], system_site_packages=True
     )
@@ -81,7 +83,7 @@ def omni_nn_cert_task():
 
         import subprocess
         from pathlib import Path
-        from airflow.models import Variable
+        from airflow.sdk import Variable
 
         tmp = subprocess.run("printenv", check=True)
         print(tmp.stdout)
@@ -103,3 +105,5 @@ def omni_nn_cert_task():
             print("Finished")
 
     omni_nn_cert_task()
+
+omni_cert_dag()