[airflow] Avoid deprecated values (AIR302) (#14582)

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

@@ -1,9 +1,14 @@
 from airflow import DAG, dag
+from airflow.timetables.simple import NullTimetable
 
 DAG(dag_id="class_default_schedule")
 
 DAG(dag_id="class_schedule", schedule="@hourly")
 
+DAG(dag_id="class_schedule_interval", schedule_interval="@hourly")
+
+DAG(dag_id="class_timetable", timetable=NullTimetable())
+
 
 @dag()
 def decorator_default_schedule():
@@ -13,3 +18,13 @@ def decorator_default_schedule():
 @dag(schedule="0 * * * *")
 def decorator_schedule():
     pass
+
+
+@dag(schedule_interval="0 * * * *")
+def decorator_schedule_interval():
+    pass
+
+
+@dag(timetable=NullTimetable())
+def decorator_timetable():
+    pass

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

@@ -0,0 +1,23 @@
+from airflow import DAG, dag
+from airflow.timetables.simple import NullTimetable
+
+DAG(dag_id="class_schedule", schedule="@hourly")
+
+DAG(dag_id="class_schedule_interval", schedule_interval="@hourly")
+
+DAG(dag_id="class_timetable", timetable=NullTimetable())
+
+
+@dag(schedule="0 * * * *")
+def decorator_schedule():
+    pass
+
+
+@dag(schedule_interval="0 * * * *")
+def decorator_schedule_interval():
+    pass
+
+
+@dag(timetable=NullTimetable())
+def decorator_timetable():
+    pass

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

@@ -0,0 +1,12 @@
+from airflow.utils import dates
+from airflow.utils.dates import date_range, datetime_to_nano, days_ago
+
+date_range
+days_ago
+
+dates.date_range
+dates.days_ago
+
+# This one was not deprecated.
+datetime_to_nano
+dates.datetime_to_nano

crates/ruff_linter/src/checkers/ast/analyze/expression.rs

@@ -220,6 +220,9 @@ pub(crate) fn expression(expr: &Expr, checker: &mut Checker) {
                     if checker.enabled(Rule::RegexFlagAlias) {
                         refurb::rules::regex_flag_alias(checker, expr);
                     }
+                    if checker.enabled(Rule::Airflow3Removal) {
+                        airflow::rules::removed_in_3(checker, expr);
+                    }
 
                     // Ex) List[...]
                     if checker.any_enabled(&[
@@ -380,6 +383,9 @@ pub(crate) fn expression(expr: &Expr, checker: &mut Checker) {
             if checker.enabled(Rule::ByteStringUsage) {
                 flake8_pyi::rules::bytestring_attribute(checker, expr);
             }
+            if checker.enabled(Rule::Airflow3Removal) {
+                airflow::rules::removed_in_3(checker, expr);
+            }
         }
         Expr::Call(
             call @ ast::ExprCall {
@@ -1084,6 +1090,9 @@ pub(crate) fn expression(expr: &Expr, checker: &mut Checker) {
             if checker.enabled(Rule::UnnecessaryRegularExpression) {
                 ruff::rules::unnecessary_regular_expression(checker, call);
             }
+            if checker.enabled(Rule::Airflow3Removal) {
+                airflow::rules::removed_in_3(checker, expr);
+            }
         }
         Expr::Dict(dict) => {
             if checker.any_enabled(&[

crates/ruff_linter/src/codes.rs

@@ -1040,6 +1040,7 @@ pub fn code_to_rule(linter: Linter, code: &str) -> Option<(RuleGroup, Rule)> {
         // airflow
         (Airflow, "001") => (RuleGroup::Stable, rules::airflow::rules::AirflowVariableNameTaskIdMismatch),
         (Airflow, "301") => (RuleGroup::Preview, rules::airflow::rules::AirflowDagNoScheduleArgument),
+        (Airflow, "302") => (RuleGroup::Preview, rules::airflow::rules::Airflow3Removal),
 
         // perflint
         (Perflint, "101") => (RuleGroup::Stable, rules::perflint::rules::UnnecessaryListCast),

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

@@ -14,6 +14,8 @@ mod tests {
 
     #[test_case(Rule::AirflowVariableNameTaskIdMismatch, Path::new("AIR001.py"))]
     #[test_case(Rule::AirflowDagNoScheduleArgument, Path::new("AIR301.py"))]
+    #[test_case(Rule::Airflow3Removal, Path::new("AIR302_args.py"))]
+    #[test_case(Rule::Airflow3Removal, Path::new("AIR302_names.py"))]
     fn rules(rule_code: Rule, path: &Path) -> Result<()> {
         let snapshot = format!("{}_{}", rule_code.noqa_code(), path.to_string_lossy());
         let diagnostics = test_path(

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

@@ -73,8 +73,14 @@ pub(crate) fn dag_no_schedule_argument(checker: &mut Checker, expr: &Expr) {
         return;
     }
 
-    // If there's a `schedule` keyword argument, we are good.
-    if arguments.find_keyword("schedule").is_some() {
+    // If there's a schedule keyword argument, we are good.
+    // This includes the canonical 'schedule', and the deprecated 'timetable'
+    // and 'schedule_interval'. Usages of deprecated schedule arguments are
+    // covered by AIR302.
+    if ["schedule", "schedule_interval", "timetable"]
+        .iter()
+        .any(|a| arguments.find_keyword(a).is_some())
+    {
         return;
     }
 

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

@@ -1,5 +1,7 @@
 pub(crate) use dag_schedule_argument::*;
+pub(crate) use removal_in_3::*;
 pub(crate) use task_variable_name::*;
 
 mod dag_schedule_argument;
+mod removal_in_3;
 mod task_variable_name;

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

@@ -0,0 +1,144 @@
+use ruff_diagnostics::{Diagnostic, Violation};
+use ruff_macros::{derive_message_formats, ViolationMetadata};
+use ruff_python_ast::{name::QualifiedName, Arguments, Expr, ExprAttribute, ExprCall};
+use ruff_python_semantic::Modules;
+use ruff_text_size::Ranged;
+
+use crate::checkers::ast::Checker;
+
+#[derive(Debug, Eq, PartialEq)]
+enum Replacement {
+    None,
+    Name(String),
+}
+
+/// ## What it does
+/// Checks for uses of deprecated Airflow functions and values.
+///
+/// ## Why is this bad?
+/// Airflow 3.0 removed various deprecated functions, members, and other
+/// values. Some have more modern replacements. Others are considered too niche
+/// and not worth to be maintained in Airflow.
+///
+/// ## Example
+/// ```python
+/// from airflow.utils.dates import days_ago
+///
+///
+/// yesterday = days_ago(today, 1)
+/// ```
+///
+/// Use instead:
+/// ```python
+/// from datetime import timedelta
+///
+///
+/// yesterday = today - timedelta(days=1)
+/// ```
+#[derive(ViolationMetadata)]
+pub(crate) struct Airflow3Removal {
+    deprecated: String,
+    replacement: Replacement,
+}
+
+impl Violation for Airflow3Removal {
+    #[derive_message_formats]
+    fn message(&self) -> String {
+        let Airflow3Removal {
+            deprecated,
+            replacement,
+        } = self;
+        match replacement {
+            Replacement::None => format!("`{deprecated}` is removed in Airflow 3.0"),
+            Replacement::Name(name) => {
+                format!("`{deprecated}` is removed in Airflow 3.0; use {name} instead")
+            }
+        }
+    }
+}
+
+fn diagnostic_for_argument(
+    arguments: &Arguments,
+    deprecated: &str,
+    replacement: Option<&str>,
+) -> Option<Diagnostic> {
+    let keyword = arguments.find_keyword(deprecated)?;
+    Some(Diagnostic::new(
+        Airflow3Removal {
+            deprecated: (*deprecated).to_string(),
+            replacement: match replacement {
+                Some(name) => Replacement::Name(name.to_owned()),
+                None => Replacement::None,
+            },
+        },
+        keyword
+            .arg
+            .as_ref()
+            .map_or_else(|| keyword.range(), Ranged::range),
+    ))
+}
+
+fn removed_argument(checker: &mut Checker, qualname: &QualifiedName, arguments: &Arguments) {
+    #[allow(clippy::single_match)]
+    match qualname.segments() {
+        ["airflow", .., "DAG" | "dag"] => {
+            checker.diagnostics.extend(diagnostic_for_argument(
+                arguments,
+                "schedule_interval",
+                Some("schedule"),
+            ));
+            checker.diagnostics.extend(diagnostic_for_argument(
+                arguments,
+                "timetable",
+                Some("schedule"),
+            ));
+        }
+        _ => {}
+    };
+}
+
+fn removed_name(checker: &mut Checker, expr: &Expr, ranged: impl Ranged) {
+    let result =
+        checker
+            .semantic()
+            .resolve_qualified_name(expr)
+            .and_then(|qualname| match qualname.segments() {
+                ["airflow", "utils", "dates", "date_range"] => {
+                    Some((qualname.to_string(), Replacement::None))
+                }
+                ["airflow", "utils", "dates", "days_ago"] => Some((
+                    qualname.to_string(),
+                    Replacement::Name("datetime.timedelta()".to_string()),
+                )),
+                _ => None,
+            });
+    if let Some((deprecated, replacement)) = result {
+        checker.diagnostics.push(Diagnostic::new(
+            Airflow3Removal {
+                deprecated,
+                replacement,
+            },
+            ranged.range(),
+        ));
+    }
+}
+
+/// AIR302
+pub(crate) fn removed_in_3(checker: &mut Checker, expr: &Expr) {
+    if !checker.semantic().seen_module(Modules::AIRFLOW) {
+        return;
+    }
+
+    match expr {
+        Expr::Call(ExprCall {
+            func, arguments, ..
+        }) => {
+            if let Some(qualname) = checker.semantic().resolve_qualified_name(func) {
+                removed_argument(checker, &qualname, arguments);
+            };
+        }
+        Expr::Attribute(ExprAttribute { attr: ranged, .. }) => removed_name(checker, expr, ranged),
+        ranged @ Expr::Name(_) => removed_name(checker, expr, ranged),
+        _ => {}
+    }
+}

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

@@ -1,20 +1,20 @@
 ---
 source: crates/ruff_linter/src/rules/airflow/mod.rs
 ---
-AIR301.py:3:1: AIR301 DAG should have an explicit `schedule` argument
+AIR301.py:4:1: AIR301 DAG should have an explicit `schedule` argument
   |
-1 | from airflow import DAG, dag
-2 | 
-3 | DAG(dag_id="class_default_schedule")
+2 | from airflow.timetables.simple import NullTimetable
+3 | 
+4 | DAG(dag_id="class_default_schedule")
   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ AIR301
-4 | 
-5 | DAG(dag_id="class_schedule", schedule="@hourly")
+5 | 
+6 | DAG(dag_id="class_schedule", schedule="@hourly")
   |
 
-AIR301.py:8:2: AIR301 DAG should have an explicit `schedule` argument
+AIR301.py:13:2: AIR301 DAG should have an explicit `schedule` argument
    |
- 8 | @dag()
+13 | @dag()
    |  ^^^^^ AIR301
- 9 | def decorator_default_schedule():
-10 |     pass
+14 | def decorator_default_schedule():
+15 |     pass
    |

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

@@ -0,0 +1,36 @@
+---
+source: crates/ruff_linter/src/rules/airflow/mod.rs
+---
+AIR302_args.py:6:39: AIR302 `schedule_interval` is removed in Airflow 3.0; use schedule instead
+  |
+4 | DAG(dag_id="class_schedule", schedule="@hourly")
+5 | 
+6 | DAG(dag_id="class_schedule_interval", schedule_interval="@hourly")
+  |                                       ^^^^^^^^^^^^^^^^^ AIR302
+7 | 
+8 | DAG(dag_id="class_timetable", timetable=NullTimetable())
+  |
+
+AIR302_args.py:8:31: AIR302 `timetable` is removed in Airflow 3.0; use schedule instead
+  |
+6 | DAG(dag_id="class_schedule_interval", schedule_interval="@hourly")
+7 | 
+8 | DAG(dag_id="class_timetable", timetable=NullTimetable())
+  |                               ^^^^^^^^^ AIR302
+  |
+
+AIR302_args.py:16:6: AIR302 `schedule_interval` is removed in Airflow 3.0; use schedule instead
+   |
+16 | @dag(schedule_interval="0 * * * *")
+   |      ^^^^^^^^^^^^^^^^^ AIR302
+17 | def decorator_schedule_interval():
+18 |     pass
+   |
+
+AIR302_args.py:21:6: AIR302 `timetable` is removed in Airflow 3.0; use schedule instead
+   |
+21 | @dag(timetable=NullTimetable())
+   |      ^^^^^^^^^ AIR302
+22 | def decorator_timetable():
+23 |     pass
+   |

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

@@ -0,0 +1,38 @@
+---
+source: crates/ruff_linter/src/rules/airflow/mod.rs
+---
+AIR302_names.py:4:1: AIR302 `airflow.utils.dates.date_range` is removed in Airflow 3.0
+  |
+2 | from airflow.utils.dates import date_range, datetime_to_nano, days_ago
+3 | 
+4 | date_range
+  | ^^^^^^^^^^ AIR302
+5 | days_ago
+  |
+
+AIR302_names.py:5:1: AIR302 `airflow.utils.dates.days_ago` is removed in Airflow 3.0; use datetime.timedelta() instead
+  |
+4 | date_range
+5 | days_ago
+  | ^^^^^^^^ AIR302
+6 | 
+7 | dates.date_range
+  |
+
+AIR302_names.py:7:7: AIR302 `airflow.utils.dates.date_range` is removed in Airflow 3.0
+  |
+5 | days_ago
+6 | 
+7 | dates.date_range
+  |       ^^^^^^^^^^ AIR302
+8 | dates.days_ago
+  |
+
+AIR302_names.py:8:7: AIR302 `airflow.utils.dates.days_ago` is removed in Airflow 3.0; use datetime.timedelta() instead
+   |
+ 7 | dates.date_range
+ 8 | dates.days_ago
+   |       ^^^^^^^^ AIR302
+ 9 | 
+10 | # This one was not deprecated.
+   |