Replace deprecated days_ago

`days_ago` is going to be deprecated and removed. Vote: https://lists.apache.org/thread/qfqjb8m3v834yc8mxo1oqtjddhp9sggk

PR: apache/airflow#21796
Issue: apache/airflow#21683

astronomer/providers/amazon/aws/example_dags/example_redshift_cluster_management.py

@@ -1,8 +1,8 @@
 import os
 
-import airflow
+from airflow.models.dag import DAG
 from airflow.operators.dummy import DummyOperator
-from airflow.utils.dates import days_ago
+from airflow.utils.timezone import datetime
 
 from astronomer.providers.amazon.aws.operators.redshift_cluster import (
     RedshiftPauseClusterOperatorAsync,
@@ -14,9 +14,9 @@
 
 REDSHIFT_CLUSTER_IDENTIFIER = os.environ.get("REDSHIFT_CLUSTER_IDENTIFIER", "astro-redshift-cluster-1")
 
-with airflow.DAG(
+with DAG(
     dag_id="example_async_redshift_cluster_management",
-    start_date=days_ago(1),
+    start_date=datetime(2021, 1, 1),
     tags=["example", "async"],
     schedule_interval="@once",
     catchup=False,

astronomer/providers/amazon/aws/example_dags/example_redshift_sql.py

@@ -1,13 +1,13 @@
-import airflow
-from airflow.utils.dates import days_ago
+from airflow.models.dag import DAG
+from airflow.utils.timezone import datetime
 
 from astronomer.providers.amazon.aws.operators.redshift_sql import (
     RedshiftSQLOperatorAsync,
 )
 
-with airflow.DAG(
+with DAG(
     dag_id="example_async_redshift_sql",
-    start_date=days_ago(1),
+    start_date=datetime(2021, 1, 1),
     tags=["example", "async"],
     schedule_interval="@once",
     catchup=False,

astronomer/providers/amazon/aws/example_dags/example_s3.py

@@ -1,7 +1,7 @@
 from datetime import timedelta
 
-from airflow import DAG
-from airflow.utils.dates import days_ago
+from airflow.models.dag import DAG
+from airflow.utils.timezone import datetime
 
 from astronomer.providers.amazon.aws.sensors.s3 import S3KeySensorAsync
 
@@ -13,7 +13,7 @@
 with DAG(
     dag_id="example_s3_key_sensor",
     schedule_interval="@daily",
-    start_date=days_ago(3),
+    start_date=datetime(2021, 1, 1),
     catchup=False,
     default_args=default_args,
     tags=["async"],

astronomer/providers/cncf/kubernetes/example_dags/example_kubernetes_pod_operator.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 
-from airflow import DAG
 from airflow.configuration import conf
+from airflow.models.dag import DAG
 
 from astronomer.providers.cncf.kubernetes.operators.kubernetes_pod import (
     KubernetesPodOperatorAsync,

astronomer/providers/core/example_dags/example_external_task.py

@@ -10,13 +10,12 @@
        successfully completes, then resumes execution and finishes without issue.
 """
 
-from airflow import DAG
-from airflow.utils.dates import days_ago
+from airflow.models.dag import DAG
+from airflow.utils.timezone import datetime
 
 from astronomer.providers.core.sensors.external_task import ExternalTaskSensorAsync
 
-default_args = {"start_date": days_ago(0)}
-with DAG("test_external_task_async", schedule_interval="@daily", default_args=default_args) as dag:
+with DAG("test_external_task_async", schedule_interval="@daily", start_date=datetime(2022, 1, 1)) as dag:
     ext_task_sensor = ExternalTaskSensorAsync(
         task_id="external_task_async",
         external_task_id="wait_for_me",

astronomer/providers/core/example_dags/example_external_task_wait_for_me.py

@@ -1,20 +1,18 @@
-import datetime
+from datetime import timedelta
 
-from airflow import DAG
+from airflow.models.dag import DAG
 from airflow.operators.dummy import DummyOperator
 from airflow.sensors.time_sensor import TimeSensorAsync
-from airflow.utils.dates import days_ago
-from airflow.utils.timezone import utcnow
+from airflow.utils.timezone import datetime, utcnow
 
-default_args = {"start_date": days_ago(0)}
 with DAG(
     "test_external_task_async_waits_for_me",
     schedule_interval="@daily",
-    default_args=default_args,
+    start_date=datetime(2022, 1, 1),
 ) as dag:
     wait_for_me = TimeSensorAsync(
         task_id="wait_for_me",
-        target_time=utcnow() + datetime.timedelta(seconds=3),
+        target_time=utcnow() + timedelta(seconds=3),
     )
     complete = DummyOperator(task_id="complete")
     wait_for_me >> complete

astronomer/providers/core/example_dags/example_file_sensor.py

@@ -1,9 +1,14 @@
-import airflow
-from airflow.utils.dates import days_ago
+from airflow.models.dag import DAG
+from airflow.utils.timezone import datetime
 
 from astronomer.providers.core.sensors.filesystem import FileSensorAsync
 
-with airflow.DAG("example_async_file_sensor", start_date=days_ago(1), tags=["async"]) as dag:
+with DAG(
+    "example_async_file_sensor",
+    start_date=datetime(2022, 1, 1),
+    catchup=False,
+    tags=["async"],
+) as dag:
     sensor_task = FileSensorAsync(
         task_id="my_file_sensor_task",
         filepath="/files/dags/example_async_file.py",

astronomer/providers/databricks/example_dags/example_databricks.py

@@ -1,7 +1,7 @@
 from datetime import timedelta
 
-from airflow import DAG
-from airflow.utils.dates import days_ago
+from airflow.models.dag import DAG
+from airflow.utils.timezone import datetime
 
 from astronomer.providers.databricks.operators.databricks import (
     DatabricksRunNowOperatorAsync,
@@ -21,7 +21,7 @@
 
 with DAG(
     "databricks_dag",
-    start_date=days_ago(0),
+    start_date=datetime(2022, 1, 1),
     schedule_interval="@daily",
     catchup=False,
     default_args=default_args,

astronomer/providers/http/example_dags/example_http.py

@@ -1,9 +1,9 @@
-from airflow import DAG
-from airflow.utils.dates import days_ago
+from airflow.models.dag import DAG
+from airflow.utils.timezone import datetime
 
 from astronomer.providers.http.sensors.http import HttpSensorAsync
 
-with DAG("example_async_http_sensor", tags=["example", "async"], start_date=days_ago(2)) as dag:
+with DAG("example_async_http_sensor", tags=["example", "async"], start_date=datetime(2022, 1, 1)) as dag:
     # This task will continue to defer as it will receive 404 every time
     async_http_sensor = HttpSensorAsync(
         task_id="async_http_sensor",

astronomer/providers/snowflake/example_dags/example_snowflake.py

@@ -18,8 +18,8 @@
 """
 Example use of SnowflakeAsync related providers.
 """
-from airflow import DAG
-from airflow.utils.dates import days_ago
+from airflow.models.dag import DAG
+from airflow.utils.timezone import datetime
 
 from astronomer.providers.snowflake.operators.snowflake import SnowflakeOperatorAsync
 
@@ -42,7 +42,7 @@
 
 dag = DAG(
     "example_snowflake",
-    start_date=days_ago(0),
+    start_date=datetime(2022, 1, 1),
     schedule_interval=None,
     default_args={"snowflake_conn_id": SNOWFLAKE_CONN_ID},
     tags=["example"],

tests/conftest.py

@@ -1,5 +1,4 @@
 import os
-from datetime import datetime
 
 import pytest
 from aioresponses import aioresponses
@@ -19,8 +18,8 @@
     XCom,
 )
 from airflow.utils import db
-from airflow.utils.dates import days_ago
 from airflow.utils.session import create_session
+from airflow.utils.timezone import datetime
 
 TEST_DAG_ID = "unit_test_dag"
 
@@ -62,8 +61,7 @@ def dag():
     """
     Creates a test DAG with default arguments.
     """
-    args = {"owner": "airflow", "start_date": days_ago(0)}
-    dag = DAG(TEST_DAG_ID, default_args=args)
+    dag = DAG(TEST_DAG_ID, start_date=datetime(2022, 1, 1))
     yield dag