[airflow] warning airflow....DAG.create_dagrun has been removed (AIR301) (#21093)

crates/ruff_linter/resources/test/fixtures/airflow/AIR301_class_attribute.py

@@ -10,6 +10,7 @@ from airflow.datasets import (
 )
 from airflow.datasets.manager import DatasetManager
 from airflow.lineage.hook import DatasetLineageInfo, HookLineageCollector
+from airflow.models.dag import DAG
 from airflow.providers.amazon.aws.auth_manager.aws_auth_manager import AwsAuthManager
 from airflow.providers.apache.beam.hooks import BeamHook, NotAir302HookError
 from airflow.providers.google.cloud.secrets.secret_manager import (
@@ -20,6 +21,7 @@ from airflow.providers_manager import ProvidersManager
 from airflow.secrets.base_secrets import BaseSecretsBackend
 from airflow.secrets.local_filesystem import LocalFilesystemBackend
 
+
 # airflow.Dataset
 dataset_from_root = DatasetFromRoot()
 dataset_from_root.iter_datasets()
@@ -56,6 +58,10 @@ hlc.add_input_dataset()
 hlc.add_output_dataset()
 hlc.collected_datasets()
 
+# airflow.models.dag.DAG
+test_dag = DAG(dag_id="test_dag")
+test_dag.create_dagrun()
+
 # airflow.providers.amazon.auth_manager.aws_auth_manager
 aam = AwsAuthManager()
 aam.is_authorized_dataset()
@@ -96,3 +102,15 @@ base_secret_backend.get_connections()
 # airflow.secrets.local_filesystem
 lfb = LocalFilesystemBackend()
 lfb.get_connections()
+
+from airflow.models import DAG
+
+# airflow.DAG
+test_dag = DAG(dag_id="test_dag")
+test_dag.create_dagrun()
+
+from airflow import DAG
+
+# airflow.DAG
+test_dag = DAG(dag_id="test_dag")
+test_dag.create_dagrun()

crates/ruff_linter/src/rules/airflow/rules/removal_in_3.rs

@@ -492,6 +492,12 @@ fn check_method(checker: &Checker, call_expr: &ExprCall) {
             "collected_datasets" => Replacement::AttrName("collected_assets"),
             _ => return,
         },
+        ["airflow", "models", "dag", "DAG"] | ["airflow", "models", "DAG"] | ["airflow", "DAG"] => {
+            match attr.as_str() {
+                "create_dagrun" => Replacement::None,
+                _ => return,
+            }
+        }
         ["airflow", "providers_manager", "ProvidersManager"] => match attr.as_str() {
             "initialize_providers_dataset_uri_resources" => {
                 Replacement::AttrName("initialize_providers_asset_uri_resources")

crates/ruff_linter/src/rules/airflow/rules/suggested_to_update_3_0.rs

@@ -288,10 +288,12 @@ fn check_name(checker: &Checker, expr: &Expr, range: TextRange) {
         },
 
         // airflow.model..DAG
-        ["airflow", "models", .., "DAG"] => Replacement::SourceModuleMoved {
+        ["airflow", "models", "dag", "DAG"] | ["airflow", "models", "DAG"] | ["airflow", "DAG"] => {
-            module: "airflow.sdk",
+            Replacement::SourceModuleMoved {
-            name: "DAG".to_string(),
+                module: "airflow.sdk",
-        },
+                name: "DAG".to_string(),
+            }
+        }
 
         // airflow.sensors.base
         [

crates/ruff_linter/src/rules/airflow/snapshots/ruff_linter__rules__airflow__tests__AIR311_AIR311_args.py.snap

@@ -1,6 +1,25 @@
 ---
 source: crates/ruff_linter/src/rules/airflow/mod.rs
 ---
+AIR311 [*] `airflow.DAG` is removed in Airflow 3.0; It still works in Airflow 3.0 but is expected to be removed in a future version.
+  --> AIR311_args.py:13:1
+   |
+13 | DAG(dag_id="class_sla_callback", sla_miss_callback=sla_callback)
+   | ^^^
+   |
+help: Use `DAG` from `airflow.sdk` instead.
+2  | 
+3  | from datetime import timedelta
+4  | 
+   - from airflow import DAG, dag
+5  + from airflow import dag
+6  | from airflow.operators.datetime import BranchDateTimeOperator
+7  + from airflow.sdk import DAG
+8  | 
+9  | 
+10 | def sla_callback(*arg, **kwargs):
+note: This is an unsafe fix and may change runtime behavior
+
 AIR311 `sla_miss_callback` is removed in Airflow 3.0; It still works in Airflow 3.0 but is expected to be removed in a future version.
   --> AIR311_args.py:13:34
    |