Upgrade ruff to latest version

.pre-commit-config.yaml

@@ -167,7 +167,7 @@ repos:
         name: Check imports in providers
         entry: ./scripts/ci/pre_commit/check_imports_in_providers.py
         language: python
-        additional_dependencies: ['rich>=12.4.4', "ruff==0.8.1"]
+        additional_dependencies: ['rich>=12.4.4', "ruff==0.11.2"]
         files: ^providers/.*/src/airflow/providers/.*\.py$
         require_serial: true
       - id: update-black-version
@@ -363,7 +363,7 @@ repos:
         types_or: [python, pyi]
         args: [--fix]
         require_serial: true
-        additional_dependencies: ["ruff==0.8.1"]
+        additional_dependencies: ["ruff==0.11.2"]
         exclude: ^airflow-core/tests/unit/dags/test_imports.py|^performance/tests/test_.*.py
       - id: ruff-format
         name: Run 'ruff format'
@@ -373,7 +373,7 @@ repos:
         types_or: [python, pyi]
         args: []
         require_serial: true
-        additional_dependencies: ["ruff==0.8.1"]
+        additional_dependencies: ["ruff==0.11.2"]
         exclude: ^airflow-core/tests/unit/dags/test_imports.py$
       - id: replace-bad-characters
         name: Replace bad characters

airflow-core/src/airflow/api_fastapi/app.py

@@ -123,8 +123,7 @@ def get_auth_manager_cls() -> type[BaseAuthManager]:
 
     if not auth_manager_cls:
         raise AirflowConfigException(
-            "No auth manager defined in the config. "
-            "Please specify one using section/key [core/auth_manager]."
+            "No auth manager defined in the config. Please specify one using section/key [core/auth_manager]."
         )
 
     return auth_manager_cls

airflow-core/src/airflow/api_fastapi/core_api/app.py

@@ -92,7 +92,7 @@ def init_plugins(app: FastAPI) -> None:
     plugins_manager.initialize_fastapi_plugins()
 
     # After calling initialize_fastapi_plugins, fastapi_apps cannot be None anymore.
-    for subapp_dict in cast(list, plugins_manager.fastapi_apps):
+    for subapp_dict in cast("list", plugins_manager.fastapi_apps):
         name = subapp_dict.get("name")
         subapp = subapp_dict.get("app")
         if subapp is None:

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_report.py

@@ -70,6 +70,6 @@ def get_dag_reports(
         filtered_dagbag_stats = []
 
     return DagReportCollectionResponse(
-        dag_reports=cast(list[DagReportResponse], filtered_dagbag_stats),
+        dag_reports=cast("list[DagReportResponse]", filtered_dagbag_stats),
         total_entries=len(filtered_dagbag_stats),
     )

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py

@@ -272,7 +272,7 @@ def clear_dag_run(
         )
 
         return TaskInstanceCollectionResponse(
-            task_instances=cast(list[TaskInstanceResponse], task_instances),
+            task_instances=cast("list[TaskInstanceResponse]", task_instances),
             total_entries=len(task_instances),
         )
     else:

airflow-core/src/airflow/api_fastapi/core_api/routes/public/plugins.py

@@ -41,6 +41,6 @@ def get_plugins(
 ) -> PluginCollectionResponse:
     plugins_info = sorted(get_plugin_info(), key=lambda x: x["name"])
     return PluginCollectionResponse(
-        plugins=cast(list[PluginResponse], plugins_info[offset.value :][: limit.value]),
+        plugins=cast("list[PluginResponse]", plugins_info[offset.value :][: limit.value]),
         total_entries=len(plugins_info),
     )

airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py

@@ -307,7 +307,7 @@ def _query(orm_object: Base) -> Select:
             f"The Task Instance with dag_id: `{dag_id}`, run_id: `{dag_run_id}`, task_id: `{task_id}` and map_index: `{map_index}` was not found",
         )
     return TaskInstanceHistoryCollectionResponse(
-        task_instances=cast(list[TaskInstanceHistoryResponse], task_instances),
+        task_instances=cast("list[TaskInstanceHistoryResponse]", task_instances),
         total_entries=len(task_instances),
     )
 

airflow-core/src/airflow/api_fastapi/core_api/routes/public/tasks.py

@@ -57,7 +57,7 @@ def get_tasks(
     except AttributeError as err:
         raise HTTPException(status.HTTP_400_BAD_REQUEST, str(err))
     return TaskCollectionResponse(
-        tasks=cast(list[TaskResponse], tasks),
+        tasks=cast("list[TaskResponse]", tasks),
         total_entries=len(tasks),
     )
 
@@ -81,4 +81,4 @@ def get_task(dag_id: str, task_id, request: Request) -> TaskResponse:
         task = dag.get_task(task_id=task_id)
     except TaskNotFound:
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"Task with id {task_id} was not found")
-    return cast(TaskResponse, task)
+    return cast("TaskResponse", task)

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/auth.py

@@ -34,6 +34,6 @@ def get_auth_links(
     menu_items = get_auth_manager().get_extra_menu_items(user=user)
 
     return MenuItemCollectionResponse(
-        menu_items=cast(list[MenuItem], menu_items),
+        menu_items=cast("list[MenuItem]", menu_items),
         total_entries=len(menu_items),
     )

airflow-core/src/airflow/api_fastapi/core_api/services/ui/connections.py

@@ -114,9 +114,9 @@ class MockAnyWidget:
         """Mock any flask appbuilder widget."""
 
     @staticmethod
-    def _get_hooks_with_mocked_fab() -> (
-        tuple[MutableMapping[str, HookInfo | None], dict[str, ConnectionFormWidgetInfo], dict[str, dict]]
-    ):
+    def _get_hooks_with_mocked_fab() -> tuple[
+        MutableMapping[str, HookInfo | None], dict[str, ConnectionFormWidgetInfo], dict[str, dict]
+    ]:
         """Get hooks with all details w/o FAB needing to be installed."""
         from unittest import mock
 

airflow-core/src/airflow/cli/commands/task_command.py

@@ -299,7 +299,7 @@ def _guess_debugger() -> _SupportedDebugger:
     exc: Exception
     for mod_name in SUPPORTED_DEBUGGER_MODULES:
         try:
-            return cast(_SupportedDebugger, importlib.import_module(mod_name))
+            return cast("_SupportedDebugger", importlib.import_module(mod_name))
         except ImportError as e:
             exc = e
     raise exc

airflow-core/src/airflow/cli/commands/variable_command.py

@@ -114,8 +114,7 @@ def variables_import(args, session):
         print(f"{fail_count} variable(s) failed to be updated.")
     if skipped:
         print(
-            f"The variables with these keys: {list(sorted(skipped))} "
-            f"were skipped because they already exists"
+            f"The variables with these keys: {list(sorted(skipped))} were skipped because they already exists"
         )
 
 

airflow-core/src/airflow/dag_processing/collection.py

@@ -51,7 +51,6 @@
 from airflow.models.errors import ParseImportError
 from airflow.models.trigger import Trigger
 from airflow.sdk.definitions.asset import Asset, AssetAlias, AssetNameRef, AssetUriRef
-from airflow.serialization.serialized_objects import SerializedAssetWatcher
 from airflow.triggers.base import BaseEventTrigger
 from airflow.utils.retries import MAX_DB_RETRIES, run_with_db_retries
 from airflow.utils.sqlalchemy import with_row_locks
@@ -65,7 +64,7 @@
     from sqlalchemy.sql import Select
 
     from airflow.models.dagwarning import DagWarning
-    from airflow.serialization.serialized_objects import MaybeSerializedDAG
+    from airflow.serialization.serialized_objects import MaybeSerializedDAG, SerializedAssetWatcher
     from airflow.typing_compat import Self
 
 log = logging.getLogger(__name__)
@@ -799,7 +798,7 @@ def add_asset_trigger_references(
         for name_uri, asset in self.assets.items():
             # If the asset belong to a DAG not active or paused, consider there is no watcher associated to it
             asset_watchers: list[SerializedAssetWatcher] = (
-                [cast(SerializedAssetWatcher, watcher) for watcher in asset.watchers]
+                [cast("SerializedAssetWatcher", watcher) for watcher in asset.watchers]
                 if name_uri in active_assets
                 else []
             )

airflow-core/src/airflow/dag_processing/manager.py

@@ -870,7 +870,7 @@ def _create_process(self, dag_file: DagFileInfo) -> DagFileProcessorProcess:
         return DagFileProcessorProcess.start(
             id=id,
             path=dag_file.absolute_path,
-            bundle_path=cast(Path, dag_file.bundle_path),
+            bundle_path=cast("Path", dag_file.bundle_path),
             callbacks=callback_to_execute_for_file,
             selector=self.selector,
             logger=logger,

airflow-core/src/airflow/decorators/base.py

@@ -492,7 +492,7 @@ def _expand(self, expand_input: ExpandInput, *, strict: bool) -> XComArg:
         partial_kwargs.setdefault("op_kwargs", {})
 
         # Mypy does not work well with a subclassed attrs class :(
-        _MappedOperator = cast(Any, DecoratedMappedOperator)
+        _MappedOperator = cast("Any", DecoratedMappedOperator)
 
         try:
             operator_name = self.operator_class.custom_operator_name  # type: ignore
@@ -665,15 +665,15 @@ def task_decorator_factory(
     it's instantiated.
     """
     if multiple_outputs is None:
-        multiple_outputs = cast(bool, attr.NOTHING)
+        multiple_outputs = cast("bool", attr.NOTHING)
     if python_callable:
         decorator = _TaskDecorator(
             function=python_callable,
             multiple_outputs=multiple_outputs,
             operator_class=decorated_operator_class,
             kwargs=kwargs,
         )
-        return cast(TaskDecorator, decorator)
+        return cast("TaskDecorator", decorator)
     elif python_callable is not None:
         raise TypeError("No args allowed while using @task, use kwargs instead")
 
@@ -685,4 +685,4 @@ def decorator_factory(python_callable):
             kwargs=kwargs,
         )
 
-    return cast(TaskDecorator, decorator_factory)
+    return cast("TaskDecorator", decorator_factory)