Replace models.BaseOperator to Task SDK one for SFTP (#52435)

* Replace models.BaseOperator to Task SDK one for SFTP

* Resolve MC, adding PokeReturnValue to version_compat.py

Author: jroachgolf84

providers/sftp/src/airflow/providers/sftp/operators/sftp.py

@@ -28,8 +28,8 @@
 import paramiko
 
 from airflow.exceptions import AirflowException
-from airflow.models import BaseOperator
 from airflow.providers.sftp.hooks.sftp import SFTPHook
+from airflow.providers.sftp.version_compat import BaseOperator
 
 
 class SFTPOperation:

providers/sftp/src/airflow/providers/sftp/sensors/sftp.py

@@ -30,14 +30,9 @@
 from airflow.exceptions import AirflowException
 from airflow.providers.sftp.hooks.sftp import SFTPHook
 from airflow.providers.sftp.triggers.sftp import SFTPTrigger
-from airflow.providers.sftp.version_compat import AIRFLOW_V_3_0_PLUS
+from airflow.providers.sftp.version_compat import BaseSensorOperator, PokeReturnValue
 from airflow.utils.timezone import convert_to_utc, parse
 
-if AIRFLOW_V_3_0_PLUS:
-    from airflow.sdk import BaseSensorOperator, PokeReturnValue
-else:
-    from airflow.sensors.base import BaseSensorOperator, PokeReturnValue  # type: ignore[no-redef]
-
 if TYPE_CHECKING:
     try:
         from airflow.sdk.definitions.context import Context

providers/sftp/src/airflow/providers/sftp/version_compat.py

@@ -33,3 +33,11 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
 
 
 AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
+
+if AIRFLOW_V_3_0_PLUS:
+    from airflow.sdk import BaseOperator, BaseSensorOperator, PokeReturnValue
+else:
+    from airflow.models import BaseOperator
+    from airflow.sensors.base import BaseSensorOperator, PokeReturnValue  # type: ignore[no-redef]
+
+__all__ = ["AIRFLOW_V_3_0_PLUS", "BaseOperator", "BaseSensorOperator", "PokeReturnValue"]