[airflow] Add unsafe fix for module moved cases (AIR312) (#18363)

<!--
Thank you for contributing to Ruff/ty! To help us out with reviewing,
please consider the following:

- Does this pull request include a summary of the change? (See below.)
- Does this pull request include a descriptive title? (Please prefix
with `[ty]` for ty pull
  requests.)
- Does this pull request include references to any relevant issues?
-->

## Summary

<!-- What's the purpose of the change? What does it do, and why? -->

Follow up on #18093 and apply it
to AIR312

## Test Plan

<!-- How was it tested? -->

The existing test fixtures have been updated

Author: Lee-W

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

@@ -7,49 +7,71 @@
 from airflow.operators.datetime import BranchDateTimeOperator
 from airflow.operators.empty import EmptyOperator
 from airflow.operators.latest_only import LatestOnlyOperator
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+from airflow.operators.weekday import BranchDayOfWeekOperator
+from airflow.sensors.date_time import DateTimeSensor
+
+FSHook()
+PackageIndexHook()
+SubprocessHook()
+
+BashOperator()
+BranchDateTimeOperator()
+TriggerDagRunOperator()
+EmptyOperator()
+
+LatestOnlyOperator()
+BranchDayOfWeekOperator()
+DateTimeSensor()
+
 from airflow.operators.python import (
     BranchPythonOperator,
     PythonOperator,
     PythonVirtualenvOperator,
     ShortCircuitOperator,
 )
-from airflow.operators.trigger_dagrun import TriggerDagRunOperator
-from airflow.operators.weekday import BranchDayOfWeekOperator
-from airflow.sensors.date_time import DateTimeSensor, DateTimeSensorAsync
+from airflow.sensors.date_time import DateTimeSensorAsync
 from airflow.sensors.external_task import (
     ExternalTaskMarker,
     ExternalTaskSensor,
-
+)
+from airflow.sensors.time_sensor import (
+    TimeSensor,
+    TimeSensorAsync,
 )
 from airflow.sensors.filesystem import FileSensor
-from airflow.sensors.time_delta import TimeDeltaSensor, TimeDeltaSensorAsync
-from airflow.sensors.time_sensor import TimeSensor, TimeSensorAsync
+
+BranchPythonOperator()
+PythonOperator()
+PythonVirtualenvOperator()
+ShortCircuitOperator()
+DateTimeSensorAsync()
+ExternalTaskMarker()
+ExternalTaskSensor()
+FileSensor()
+TimeSensor()
+TimeSensorAsync()
+
+from airflow.sensors.time_delta import (
+    TimeDeltaSensor,
+    TimeDeltaSensorAsync,
+)
 from airflow.sensors.weekday import DayOfWeekSensor
-from airflow.triggers.external_task import DagStateTrigger, WorkflowTrigger
+from airflow.triggers.external_task import (
+    DagStateTrigger,
+    WorkflowTrigger,
+)
 from airflow.triggers.file import FileTrigger
-from airflow.triggers.temporal import DateTimeTrigger, TimeDeltaTrigger
-
-FSHook()
-PackageIndexHook()
-SubprocessHook()
-BashOperator()
-BranchDateTimeOperator()
-TriggerDagRunOperator()
-EmptyOperator()
-LatestOnlyOperator()
-(
-    BranchPythonOperator(),
-    PythonOperator(),
-    PythonVirtualenvOperator(),
-    ShortCircuitOperator(),
+from airflow.triggers.temporal import (
+    DateTimeTrigger,
+    TimeDeltaTrigger,
 )
-BranchDayOfWeekOperator()
-DateTimeSensor(), DateTimeSensorAsync()
-ExternalTaskMarker(), ExternalTaskSensor()
-FileSensor()
-TimeSensor(), TimeSensorAsync()
-TimeDeltaSensor(), TimeDeltaSensorAsync()
+
+TimeDeltaSensor()
+TimeDeltaSensorAsync()
 DayOfWeekSensor()
-DagStateTrigger(), WorkflowTrigger()
+DagStateTrigger()
+WorkflowTrigger()
 FileTrigger()
-DateTimeTrigger(), TimeDeltaTrigger()
+DateTimeTrigger()
+TimeDeltaTrigger()

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

@@ -1,16 +1,16 @@
-use crate::importer::ImportRequest;
-
-use crate::rules::airflow::helpers::{ProviderReplacement, is_guarded_by_try_except};
-use crate::{Edit, Fix, FixAvailability, Violation};
+use crate::checkers::ast::Checker;
+use crate::rules::airflow::helpers::{
+    ProviderReplacement, generate_import_edit, generate_remove_and_runtime_import_edit,
+    is_guarded_by_try_except,
+};
+use crate::{FixAvailability, Violation};
 use ruff_macros::{ViolationMetadata, derive_message_formats};
 use ruff_python_ast::name::QualifiedName;
 use ruff_python_ast::{Expr, ExprAttribute};
 use ruff_python_semantic::Modules;
 use ruff_text_size::Ranged;
 use ruff_text_size::TextRange;
 
-use crate::checkers::ast::Checker;
-
 /// ## What it does
 /// Checks for uses of Airflow functions and values that have been moved to its providers
 /// but still have a compatibility layer (e.g., `apache-airflow-providers-standard`).
@@ -302,22 +302,17 @@ fn check_names_moved_to_provider(checker: &Checker, expr: &Expr, ranged: TextRan
     if is_guarded_by_try_except(expr, module, name, checker.semantic()) {
         return;
     }
+    let mut diagnostic = checker.report_diagnostic(
+        Airflow3SuggestedToMoveToProvider {
+            deprecated: qualified_name,
+            replacement: replacement.clone(),
+        },
+        ranged,
+    );
 
-    checker
-        .report_diagnostic(
-            Airflow3SuggestedToMoveToProvider {
-                deprecated: qualified_name,
-                replacement: replacement.clone(),
-            },
-            ranged.range(),
-        )
-        .try_set_fix(|| {
-            let (import_edit, binding) = checker.importer().get_or_import_symbol(
-                &ImportRequest::import_from(module, name),
-                expr.start(),
-                checker.semantic(),
-            )?;
-            let replacement_edit = Edit::range_replacement(binding, ranged.range());
-            Ok(Fix::safe_edits(import_edit, [replacement_edit]))
-        });
+    if let Some(fix) = generate_import_edit(expr, checker, module, name, ranged)
+        .or_else(|| generate_remove_and_runtime_import_edit(expr, checker, module, name))
+    {
+        diagnostic.set_fix(fix);
+    }
 }