[airflow] Add unsafe fix module moved cases (AIR302) (#18093)

<!--
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? -->

Add utility functions `generate_import_edit` and
`generate_remove_and_runtime_import_edit` to generate the fix needed for
the airflow rules.

1. `generate_import_edit` is for the cases where the member name has
changed. (e.g., `airflow.datasts.Dataset` to `airflow.sdk.Asset`) It's
just extracted from the original logic
2. `generate_remove_and_runtime_import_edit` is for cases where the
member name has not changed. (e.g.,
`airflow.operators.pig_operator.PigOperator` to
`airflow.providers.apache.pig.hooks.pig.PigCliHook`) This is newly
introduced. As it introduced runtime import, I mark it as an unsafe fix.
Under the hook, it tried to find the original import statement, remove
it, and add a new import fix

---

* rules fix
* `airflow.sensors.external_task_sensor.ExternalTaskSensorLink` →
`airflow.providers.standard.sensors.external_task.ExternalDagLink`

## Test Plan

<!-- How was it tested? -->
The existing test fixtures have been updated

Author: Lee-W

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

@@ -5,35 +5,30 @@
     provide_bucket_name,
 )
 from airflow.operators.gcs_to_s3 import GCSToS3Operator
-from airflow.operators.google_api_to_s3_transfer import (
-    GoogleApiToS3Operator,
-    GoogleApiToS3Transfer,
-)
-from airflow.operators.redshift_to_s3_operator import (
-    RedshiftToS3Operator,
-    RedshiftToS3Transfer,
-)
+from airflow.operators.google_api_to_s3_transfer import GoogleApiToS3Operator
+from airflow.operators.redshift_to_s3_operator import RedshiftToS3Operator
 from airflow.operators.s3_file_transform_operator import S3FileTransformOperator
-from airflow.operators.s3_to_redshift_operator import (
-    S3ToRedshiftOperator,
-    S3ToRedshiftTransfer,
-)
+from airflow.operators.s3_to_redshift_operator import S3ToRedshiftOperator
 from airflow.sensors.s3_key_sensor import S3KeySensor
 
 S3Hook()
 provide_bucket_name()
 
 GCSToS3Operator()
-
 GoogleApiToS3Operator()
+RedshiftToS3Operator()
+S3FileTransformOperator()
+S3ToRedshiftOperator()
+S3KeySensor()
+
+from airflow.operators.google_api_to_s3_transfer import GoogleApiToS3Transfer
+
 GoogleApiToS3Transfer()
 
-RedshiftToS3Operator()
+from airflow.operators.redshift_to_s3_operator import RedshiftToS3Transfer
+
 RedshiftToS3Transfer()
 
-S3FileTransformOperator()
+from airflow.operators.s3_to_redshift_operator import S3ToRedshiftTransfer
 
-S3ToRedshiftOperator()
 S3ToRedshiftTransfer()
-
-S3KeySensor()

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

@@ -4,10 +4,13 @@
     ConnectorProtocol,
     DbApiHook,
 )
+
+ConnectorProtocol()
+DbApiHook()
+
 from airflow.hooks.dbapi_hook import DbApiHook
 from airflow.operators.check_operator import SQLCheckOperator
 
-ConnectorProtocol()
 DbApiHook()
 SQLCheckOperator()
 

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

@@ -12,55 +12,59 @@
 )
 from airflow.operators.hive_operator import HiveOperator
 from airflow.operators.hive_stats_operator import HiveStatsCollectionOperator
-from airflow.operators.hive_to_mysql import (
-    HiveToMySqlOperator,
-    HiveToMySqlTransfer,
-)
+from airflow.operators.hive_to_mysql import HiveToMySqlOperator
 from airflow.operators.hive_to_samba_operator import HiveToSambaOperator
-from airflow.operators.mssql_to_hive import (
-    MsSqlToHiveOperator,
-    MsSqlToHiveTransfer,
-)
-from airflow.operators.mysql_to_hive import (
-    MySqlToHiveOperator,
-    MySqlToHiveTransfer,
-)
-from airflow.operators.s3_to_hive_operator import (
-    S3ToHiveOperator,
-    S3ToHiveTransfer,
-)
-from airflow.sensors.hive_partition_sensor import HivePartitionSensor
-from airflow.sensors.metastore_partition_sensor import MetastorePartitionSensor
-from airflow.sensors.named_hive_partition_sensor import NamedHivePartitionSensor
-
-closest_ds_partition()
-max_partition()
 
+HIVE_QUEUE_PRIORITIES
 HiveCliHook()
 HiveMetastoreHook()
 HiveServer2Hook()
-HIVE_QUEUE_PRIORITIES
 
-HiveOperator()
+closest_ds_partition()
+max_partition()
 
+HiveOperator()
 HiveStatsCollectionOperator()
-
 HiveToMySqlOperator()
+HiveToSambaOperator()
+
+
+from airflow.operators.hive_to_mysql import HiveToMySqlTransfer
+
 HiveToMySqlTransfer()
 
-HiveToSambaOperator()
+from airflow.operators.mysql_to_hive import MySqlToHiveOperator
+
+MySqlToHiveOperator()
+
+from airflow.operators.mysql_to_hive import MySqlToHiveTransfer
+
+MySqlToHiveTransfer()
+
+from airflow.operators.mssql_to_hive import MsSqlToHiveOperator
 
 MsSqlToHiveOperator()
+
+from airflow.operators.mssql_to_hive import MsSqlToHiveTransfer
+
 MsSqlToHiveTransfer()
 
-MySqlToHiveOperator()
-MySqlToHiveTransfer()
+from airflow.operators.s3_to_hive_operator import S3ToHiveOperator
 
 S3ToHiveOperator()
+
+from airflow.operators.s3_to_hive_operator import S3ToHiveTransfer
+
 S3ToHiveTransfer()
 
+from airflow.sensors.hive_partition_sensor import HivePartitionSensor
+
 HivePartitionSensor()
 
+from airflow.sensors.metastore_partition_sensor import MetastorePartitionSensor
+
 MetastorePartitionSensor()
 
+from airflow.sensors.named_hive_partition_sensor import NamedHivePartitionSensor
+
 NamedHivePartitionSensor()

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

@@ -16,14 +16,7 @@
 from airflow.kubernetes.kubernetes_helper_functions import (
     add_pod_suffix,
     annotations_for_logging_task_metadata,
-    annotations_to_key,
     create_pod_id,
-    get_logs_task_metadata,
-    rand_str,
-)
-from airflow.kubernetes.pod import (
-    Port,
-    Resources,
 )
 
 ALL_NAMESPACES
@@ -37,21 +30,13 @@
 get_kube_client()
 
 add_pod_suffix()
-create_pod_id()
-
 annotations_for_logging_task_metadata()
-annotations_to_key()
-get_logs_task_metadata()
-rand_str()
-
-Port()
-Resources()
+create_pod_id()
 
 
 from airflow.kubernetes.pod_generator import (
     PodDefaults,
     PodGenerator,
-    PodGeneratorDeprecated,
     add_pod_suffix,
     datetime_to_label_safe_datestring,
     extend_object_field,
@@ -61,18 +46,16 @@
     rand_str,
 )
 
+PodDefaults()
+PodGenerator()
+add_pod_suffix()
 datetime_to_label_safe_datestring()
 extend_object_field()
 label_safe_datestring_to_datetime()
 make_safe_label_value()
 merge_objects()
-PodGenerator()
-PodDefaults()
-PodGeneratorDeprecated()
-add_pod_suffix()
 rand_str()
 
-
 from airflow.kubernetes.pod_generator_deprecated import (
     PodDefaults,
     PodGenerator,
@@ -90,7 +73,6 @@
 PodLauncher()
 PodStatus()
 
-
 from airflow.kubernetes.pod_launcher_deprecated import (
     PodDefaults,
     PodLauncher,
@@ -115,3 +97,17 @@
 Secret()
 Volume()
 VolumeMount()
+
+from airflow.kubernetes.kubernetes_helper_functions import (
+    annotations_to_key,
+    get_logs_task_metadata,
+    rand_str,
+)
+
+annotations_to_key()
+get_logs_task_metadata()
+rand_str()
+
+from airflow.kubernetes.pod_generator import PodGeneratorDeprecated
+
+PodGeneratorDeprecated()

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

@@ -5,10 +5,6 @@
     TriggerDagRunLink,
     TriggerDagRunOperator,
 )
-from airflow.operators.dummy import (
-    DummyOperator,
-    EmptyOperator,
-)
 from airflow.operators.latest_only_operator import LatestOnlyOperator
 from airflow.operators.python_operator import (
     BranchPythonOperator,
@@ -19,15 +15,12 @@
 from airflow.sensors.external_task_sensor import (
     ExternalTaskMarker,
     ExternalTaskSensor,
-    ExternalTaskSensorLink,
 )
 
 BashOperator()
 
 TriggerDagRunLink()
 TriggerDagRunOperator()
-DummyOperator()
-EmptyOperator()
 
 LatestOnlyOperator()
 
@@ -38,25 +31,48 @@
 
 ExternalTaskMarker()
 ExternalTaskSensor()
-ExternalTaskSensorLink()
 
-from airflow.operators.dummy_operator import (
-    DummyOperator,
-    EmptyOperator,
-)
-
-DummyOperator()
-EmptyOperator()
 
 from airflow.hooks.subprocess import SubprocessResult
+
 SubprocessResult()
+
 from airflow.hooks.subprocess import working_directory
+
 working_directory()
+
 from airflow.operators.datetime import target_times_as_dates
+
 target_times_as_dates()
+
 from airflow.operators.trigger_dagrun import TriggerDagRunLink
+
 TriggerDagRunLink()
+
 from airflow.sensors.external_task import ExternalTaskSensorLink
+
 ExternalTaskSensorLink()
+
 from airflow.sensors.time_delta import WaitSensor
-WaitSensor()
+
+WaitSensor()
+
+from airflow.operators.dummy import DummyOperator
+
+DummyOperator()
+
+from airflow.operators.dummy import EmptyOperator
+
+EmptyOperator()
+
+from airflow.operators.dummy_operator import DummyOperator
+
+DummyOperator()
+
+from airflow.operators.dummy_operator import EmptyOperator
+
+EmptyOperator()
+
+from airflow.sensors.external_task_sensor import ExternalTaskSensorLink
+
+ExternalTaskSensorLink()