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

## Summary

This is a follow up PR to astral-sh/ruff#21096

The new code AIR303 is added for checking function signature change in
Airflow 3.0. The new rule added to AIR303 will check if positional
argument is passed into
`airflow.lineage.hook.HookLineageCollector.create_asset`. Since this
method is updated to accept only keywords argument, passing positional
argument into it is not allowed, and will raise an error. The test is
done by checking whether positional argument with 0 index can be found.

## Test Plan

A new test file is added to the fixtures for the code AIR303. Snapshot
test is updated accordingly.

<img width="1444" height="513" alt="Screenshot from 2025-12-17 20-54-48"
src="https://github.com/user-attachments/assets/bc235195-e986-4743-9bf7-bba65805fb87"
/>

<img width="981" height="433" alt="Screenshot from 2025-12-17 21-34-29"
src="https://github.com/user-attachments/assets/492db71f-58f2-40ba-ad2f-f74852fa5a6b"
/>

Author: sjyangkevin

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

@@ -1124,6 +1124,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=...)`)