[airflow] Passing positional argument into airflow.lineage.hook.HookLineageCollector.create_asset is not allowed (AIR303)

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

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from airflow.lineage.hook import HookLineageCollector
+
+# airflow.lineage.hook
+hlc = HookLineageCollector()
+hlc.create_asset("there")
+hlc.create_asset("should", "be", "no", "posarg")
+hlc.create_asset(name="but", uri="kwargs are ok")
+hlc.create_asset()
+
+HookLineageCollector().create_asset(name="but", uri="kwargs are ok")
+HookLineageCollector().create_asset("there")
+HookLineageCollector().create_asset("should", "be", "no", "posarg")
+
+args = ["uri_value"]
+hlc.create_asset(*args)
+HookLineageCollector().create_asset(*args)
+
+# Literal unpacking
+hlc.create_asset(*["literal_uri"])
+HookLineageCollector().create_asset(*["literal_uri"])
+
+# starred args with keyword args
+hlc.create_asset(*args, extra="value")
+HookLineageCollector().create_asset(*args, extra="value")
+
+# Double-starred keyword arguments
+kwargs = {"uri": "value", "name": "test"}
+hlc.create_asset(**kwargs)
+HookLineageCollector().create_asset(**kwargs)

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

@@ -1278,6 +1278,9 @@ pub(crate) fn expression(expr: &Expr, checker: &Checker) {
             if checker.is_rule_enabled(Rule::Airflow3SuggestedUpdate) {
                 airflow::rules::airflow_3_0_suggested_update_expr(checker, expr);
             }
+            if checker.is_rule_enabled(Rule::Airflow3IncompatibleFunctionSignature) {
+                airflow::rules::airflow_3_incompatible_function_signature(checker, expr);
+            }
             if checker.is_rule_enabled(Rule::UnnecessaryCastToInt) {
                 ruff::rules::unnecessary_cast_to_int(checker, call);
             }

crates/ruff_linter/src/codes.rs

@@ -1123,6 +1123,7 @@ pub fn code_to_rule(linter: Linter, code: &str) -> Option<(RuleGroup, Rule)> {
         (Airflow, "002") => rules::airflow::rules::AirflowDagNoScheduleArgument,
         (Airflow, "301") => rules::airflow::rules::Airflow3Removal,
         (Airflow, "302") => rules::airflow::rules::Airflow3MovedToProvider,
+        (Airflow, "303") => rules::airflow::rules::Airflow3IncompatibleFunctionSignature,
         (Airflow, "311") => rules::airflow::rules::Airflow3SuggestedUpdate,
         (Airflow, "312") => rules::airflow::rules::Airflow3SuggestedToMoveToProvider,
 

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

@@ -47,6 +47,7 @@ mod tests {
     #[test_case(Rule::Airflow3MovedToProvider, Path::new("AIR302_zendesk.py"))]
     #[test_case(Rule::Airflow3MovedToProvider, Path::new("AIR302_standard.py"))]
     #[test_case(Rule::Airflow3MovedToProvider, Path::new("AIR302_try.py"))]
+    #[test_case(Rule::Airflow3IncompatibleFunctionSignature, Path::new("AIR303.py"))]
     #[test_case(Rule::Airflow3SuggestedUpdate, Path::new("AIR311_args.py"))]
     #[test_case(Rule::Airflow3SuggestedUpdate, Path::new("AIR311_names.py"))]
     #[test_case(Rule::Airflow3SuggestedUpdate, Path::new("AIR311_try.py"))]

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

@@ -0,0 +1,128 @@
+use crate::checkers::ast::Checker;
+use crate::{FixAvailability, Violation};
+use ruff_macros::{ViolationMetadata, derive_message_formats};
+use ruff_python_ast::name::QualifiedName;
+use ruff_python_ast::{Arguments, Expr, ExprAttribute, ExprCall, Identifier};
+use ruff_python_semantic::Modules;
+use ruff_python_semantic::analyze::typing;
+use ruff_text_size::Ranged;
+
+/// ## What it does
+/// Checks for Airflow function calls that will raise a runtime error in Airflow 3.0
+/// due to function signature changes, such as functions that changed to accept only
+/// keyword arguments, parameter reordering, or parameter type changes.
+///
+/// ## Why is this bad?
+/// Airflow 3.0 introduces changes to function signatures. Code that
+/// worked in Airflow 2.x will raise a runtime error if not updated in Airflow
+/// 3.0.
+///
+/// ## Example
+/// ```python
+/// from airflow.lineage.hook import HookLineageCollector
+///
+/// collector = HookLineageCollector()
+/// # Passing positional arguments will raise a runtime error in Airflow 3.0
+/// collector.create_asset("s3://bucket/key")
+/// ```
+///
+/// Use instead:
+/// ```python
+/// from airflow.lineage.hook import HookLineageCollector
+///
+/// collector = HookLineageCollector()
+/// # Passing arguments as keyword arguments instead of positional arguments
+/// collector.create_asset(uri="s3://bucket/key")
+/// ```
+#[derive(ViolationMetadata)]
+#[violation_metadata(preview_since = "0.14.11")]
+pub(crate) struct Airflow3IncompatibleFunctionSignature {
+    function_name: String,
+    change_type: FunctionSignatureChangeType,
+}
+
+impl Violation for Airflow3IncompatibleFunctionSignature {
+    const FIX_AVAILABILITY: FixAvailability = FixAvailability::None;
+
+    #[derive_message_formats]
+    fn message(&self) -> String {
+        let Airflow3IncompatibleFunctionSignature {
+            function_name,
+            change_type,
+        } = self;
+        match change_type {
+            FunctionSignatureChangeType::KeywordOnly { .. } => {
+                format!("`{function_name}` signature is changed in Airflow 3.0")
+            }
+        }
+    }
+
+    fn fix_title(&self) -> Option<String> {
+        let Airflow3IncompatibleFunctionSignature { change_type, .. } = self;
+        match change_type {
+            FunctionSignatureChangeType::KeywordOnly { message } => Some(message.to_string()),
+        }
+    }
+}
+
+/// AIR303
+pub(crate) fn airflow_3_incompatible_function_signature(checker: &Checker, expr: &Expr) {
+    if !checker.semantic().seen_module(Modules::AIRFLOW) {
+        return;
+    }
+
+    let Expr::Call(ExprCall {
+        func, arguments, ..
+    }) = expr
+    else {
+        return;
+    };
+
+    let Expr::Attribute(ExprAttribute { attr, value, .. }) = func.as_ref() else {
+        return;
+    };
+
+    // Resolve the qualified name: try variable assignments first, then fall back to direct
+    // constructor calls.
+    let qualified_name = typing::resolve_assignment(value, checker.semantic()).or_else(|| {
+        value
+            .as_call_expr()
+            .and_then(|call| checker.semantic().resolve_qualified_name(&call.func))
+    });
+
+    let Some(qualified_name) = qualified_name else {
+        return;
+    };
+
+    check_keyword_only_method(checker, &qualified_name, attr, arguments);
+}
+
+fn check_keyword_only_method(
+    checker: &Checker,
+    qualified_name: &QualifiedName,
+    attr: &Identifier,
+    arguments: &Arguments,
+) {
+    let has_positional_args =
+        arguments.find_positional(0).is_some() || arguments.args.iter().any(Expr::is_starred_expr);
+
+    if let ["airflow", "lineage", "hook", "HookLineageCollector"] = qualified_name.segments() {
+        if attr.as_str() == "create_asset" && has_positional_args {
+            checker.report_diagnostic(
+                Airflow3IncompatibleFunctionSignature {
+                    function_name: attr.to_string(),
+                    change_type: FunctionSignatureChangeType::KeywordOnly {
+                        message: "Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)",
+                    },
+                },
+                attr.range(),
+            );
+        }
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub(crate) enum FunctionSignatureChangeType {
+    /// Function signature changed to only accept keyword arguments.
+    KeywordOnly { message: &'static str },
+}

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

@@ -1,11 +1,13 @@
 pub(crate) use dag_schedule_argument::*;
+pub(crate) use function_signature_change_in_3::*;
 pub(crate) use moved_to_provider_in_3::*;
 pub(crate) use removal_in_3::*;
 pub(crate) use suggested_to_move_to_provider_in_3::*;
 pub(crate) use suggested_to_update_3_0::*;
 pub(crate) use task_variable_name::*;
 
 mod dag_schedule_argument;
+mod function_signature_change_in_3;
 mod moved_to_provider_in_3;
 mod removal_in_3;
 mod suggested_to_move_to_provider_in_3;

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

@@ -0,0 +1,114 @@
+---
+source: crates/ruff_linter/src/rules/airflow/mod.rs
+---
+AIR303 `create_asset` signature is changed in Airflow 3.0
+ --> AIR303.py:7:5
+  |
+5 | # airflow.lineage.hook
+6 | hlc = HookLineageCollector()
+7 | hlc.create_asset("there")
+  |     ^^^^^^^^^^^^
+8 | hlc.create_asset("should", "be", "no", "posarg")
+9 | hlc.create_asset(name="but", uri="kwargs are ok")
+  |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)
+
+AIR303 `create_asset` signature is changed in Airflow 3.0
+  --> AIR303.py:8:5
+   |
+ 6 | hlc = HookLineageCollector()
+ 7 | hlc.create_asset("there")
+ 8 | hlc.create_asset("should", "be", "no", "posarg")
+   |     ^^^^^^^^^^^^
+ 9 | hlc.create_asset(name="but", uri="kwargs are ok")
+10 | hlc.create_asset()
+   |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)
+
+AIR303 `create_asset` signature is changed in Airflow 3.0
+  --> AIR303.py:13:24
+   |
+12 | HookLineageCollector().create_asset(name="but", uri="kwargs are ok")
+13 | HookLineageCollector().create_asset("there")
+   |                        ^^^^^^^^^^^^
+14 | HookLineageCollector().create_asset("should", "be", "no", "posarg")
+   |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)
+
+AIR303 `create_asset` signature is changed in Airflow 3.0
+  --> AIR303.py:14:24
+   |
+12 | HookLineageCollector().create_asset(name="but", uri="kwargs are ok")
+13 | HookLineageCollector().create_asset("there")
+14 | HookLineageCollector().create_asset("should", "be", "no", "posarg")
+   |                        ^^^^^^^^^^^^
+15 |
+16 | args = ["uri_value"]
+   |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)
+
+AIR303 `create_asset` signature is changed in Airflow 3.0
+  --> AIR303.py:17:5
+   |
+16 | args = ["uri_value"]
+17 | hlc.create_asset(*args)
+   |     ^^^^^^^^^^^^
+18 | HookLineageCollector().create_asset(*args)
+   |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)
+
+AIR303 `create_asset` signature is changed in Airflow 3.0
+  --> AIR303.py:18:24
+   |
+16 | args = ["uri_value"]
+17 | hlc.create_asset(*args)
+18 | HookLineageCollector().create_asset(*args)
+   |                        ^^^^^^^^^^^^
+19 |
+20 | # Literal unpacking
+   |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)
+
+AIR303 `create_asset` signature is changed in Airflow 3.0
+  --> AIR303.py:21:5
+   |
+20 | # Literal unpacking
+21 | hlc.create_asset(*["literal_uri"])
+   |     ^^^^^^^^^^^^
+22 | HookLineageCollector().create_asset(*["literal_uri"])
+   |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)
+
+AIR303 `create_asset` signature is changed in Airflow 3.0
+  --> AIR303.py:22:24
+   |
+20 | # Literal unpacking
+21 | hlc.create_asset(*["literal_uri"])
+22 | HookLineageCollector().create_asset(*["literal_uri"])
+   |                        ^^^^^^^^^^^^
+23 |
+24 | # starred args with keyword args
+   |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)
+
+AIR303 `create_asset` signature is changed in Airflow 3.0
+  --> AIR303.py:25:5
+   |
+24 | # starred args with keyword args
+25 | hlc.create_asset(*args, extra="value")
+   |     ^^^^^^^^^^^^
+26 | HookLineageCollector().create_asset(*args, extra="value")
+   |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)
+
+AIR303 `create_asset` signature is changed in Airflow 3.0
+  --> AIR303.py:26:24
+   |
+24 | # starred args with keyword args
+25 | hlc.create_asset(*args, extra="value")
+26 | HookLineageCollector().create_asset(*args, extra="value")
+   |                        ^^^^^^^^^^^^
+27 |
+28 | # Double-starred keyword arguments
+   |
+help: Pass positional arguments as keyword arguments (e.g., `create_asset(uri=...)`)