Convert airflow/hooks to __future__.annotations

airflow/hooks/__init__.py

@@ -15,8 +15,10 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-# fmt: off
+# fmt:, off
 """Hooks."""
+from __future__ import annotations
+
 from airflow.utils.deprecation_tools import add_deprecated_classes
 
 __deprecated_classes = {

airflow/hooks/base.py

@@ -16,9 +16,11 @@
 # specific language governing permissions and limitations
 # under the License.
 """Base class for all hooks"""
+from __future__ import annotations
+
 import logging
 import warnings
-from typing import TYPE_CHECKING, Any, Dict, List
+from typing import TYPE_CHECKING, Any
 
 from airflow.exceptions import RemovedInAirflow3Warning
 from airflow.typing_compat import Protocol
@@ -40,7 +42,7 @@ class BaseHook(LoggingMixin):
     """
 
     @classmethod
-    def get_connections(cls, conn_id: str) -> List["Connection"]:
+    def get_connections(cls, conn_id: str) -> list[Connection]:
         """
         Get all connections as an iterable, given the connection id.
 
@@ -56,7 +58,7 @@ def get_connections(cls, conn_id: str) -> List["Connection"]:
         return [cls.get_connection(conn_id)]
 
     @classmethod
-    def get_connection(cls, conn_id: str) -> "Connection":
+    def get_connection(cls, conn_id: str) -> Connection:
         """
         Get connection, given connection id.
 
@@ -70,7 +72,7 @@ def get_connection(cls, conn_id: str) -> "Connection":
         return conn
 
     @classmethod
-    def get_hook(cls, conn_id: str) -> "BaseHook":
+    def get_hook(cls, conn_id: str) -> BaseHook:
         """
         Returns default hook for this connection id.
 
@@ -87,11 +89,11 @@ def get_conn(self) -> Any:
         raise NotImplementedError()
 
     @classmethod
-    def get_connection_form_widgets(cls) -> Dict[str, Any]:
+    def get_connection_form_widgets(cls) -> dict[str, Any]:
         ...
 
     @classmethod
-    def get_ui_field_behaviour(cls) -> Dict[str, Any]:
+    def get_ui_field_behaviour(cls) -> dict[str, Any]:
         ...
 
 
@@ -140,7 +142,7 @@ def get_ui_field_behaviour(cls):
     hook_name: str
 
     @staticmethod
-    def get_connection_form_widgets() -> Dict[str, Any]:
+    def get_connection_form_widgets() -> dict[str, Any]:
         """
         Returns dictionary of widgets to be added for the hook to handle extra values.
 
@@ -156,7 +158,7 @@ def get_connection_form_widgets() -> Dict[str, Any]:
         ...
 
     @staticmethod
-    def get_ui_field_behaviour() -> Dict[str, Any]:
+    def get_ui_field_behaviour() -> dict[str, Any]:
         """
         Returns dictionary describing customizations to implement in javascript handling the
         connection form. Should be compliant with airflow/customized_form_field_behaviours.schema.json'

airflow/hooks/dbapi.py

@@ -15,6 +15,8 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from __future__ import annotations
+
 import warnings
 
 from airflow.exceptions import RemovedInAirflow3Warning

airflow/hooks/filesystem.py

@@ -15,7 +15,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-#
+from __future__ import annotations
 
 from airflow.hooks.base import BaseHook
 

airflow/hooks/subprocess.py

@@ -14,13 +14,14 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from __future__ import annotations
+
 import contextlib
 import os
 import signal
 from collections import namedtuple
 from subprocess import PIPE, STDOUT, Popen
 from tempfile import TemporaryDirectory, gettempdir
-from typing import Dict, List, Optional
 
 from airflow.hooks.base import BaseHook
 
@@ -31,15 +32,15 @@ class SubprocessHook(BaseHook):
     """Hook for running processes with the ``subprocess`` module"""
 
     def __init__(self) -> None:
-        self.sub_process: Optional[Popen[bytes]] = None
+        self.sub_process: Popen[bytes] | None = None
         super().__init__()
 
     def run_command(
         self,
-        command: List[str],
-        env: Optional[Dict[str, str]] = None,
+        command: list[str],
+        env: dict[str, str] | None = None,
         output_encoding: str = 'utf-8',
-        cwd: Optional[str] = None,
+        cwd: str | None = None,
     ) -> SubprocessResult:
         """
         Execute the command.