Convert airflow/cli to __future__.annotations

airflow/cli/cli_parser.py

@@ -17,14 +17,15 @@
 # specific language governing permissions and limitations
 # under the License.
 """Command-line interface"""
+from __future__ import annotations
 
 import argparse
 import json
 import os
 import textwrap
 from argparse import Action, ArgumentError, RawTextHelpFormatter
 from functools import lru_cache
-from typing import Callable, Dict, Iterable, List, NamedTuple, Optional, Union
+from typing import Callable, Iterable, NamedTuple, Union
 
 import lazy_object_proxy
 
@@ -960,8 +961,8 @@ class ActionCommand(NamedTuple):
     help: str
     func: Callable
     args: Iterable[Arg]
-    description: Optional[str] = None
-    epilog: Optional[str] = None
+    description: str | None = None
+    epilog: str | None = None
 
 
 class GroupCommand(NamedTuple):
@@ -970,8 +971,8 @@ class GroupCommand(NamedTuple):
     name: str
     help: str
     subcommands: Iterable
-    description: Optional[str] = None
-    epilog: Optional[str] = None
+    description: str | None = None
+    epilog: str | None = None
 
 
 CLICommand = Union[ActionCommand, GroupCommand]
@@ -1801,7 +1802,7 @@ class GroupCommand(NamedTuple):
     ),
 )
 
-airflow_commands: List[CLICommand] = [
+airflow_commands: list[CLICommand] = [
     GroupCommand(
         name='dags',
         help='Manage DAGs',
@@ -1999,7 +2000,7 @@ class GroupCommand(NamedTuple):
         args=tuple(),
     ),
 ]
-ALL_COMMANDS_DICT: Dict[str, CLICommand] = {sp.name: sp for sp in airflow_commands}
+ALL_COMMANDS_DICT: dict[str, CLICommand] = {sp.name: sp for sp in airflow_commands}
 
 
 def _remove_dag_id_opt(command: ActionCommand):
@@ -2008,7 +2009,7 @@ def _remove_dag_id_opt(command: ActionCommand):
     return ActionCommand(**cmd)
 
 
-dag_cli_commands: List[CLICommand] = [
+dag_cli_commands: list[CLICommand] = [
     GroupCommand(
         name='dags',
         help='Manage DAGs',
@@ -2024,7 +2025,7 @@ def _remove_dag_id_opt(command: ActionCommand):
         subcommands=[_remove_dag_id_opt(sp) for sp in TASKS_COMMANDS if sp.name in ['list', 'test', 'run']],
     ),
 ]
-DAG_CLI_DICT: Dict[str, CLICommand] = {sp.name: sp for sp in dag_cli_commands}
+DAG_CLI_DICT: dict[str, CLICommand] = {sp.name: sp for sp in dag_cli_commands}
 
 
 class AirflowHelpFormatter(argparse.HelpFormatter):

airflow/cli/commands/celery_command.py

@@ -16,9 +16,9 @@
 # specific language governing permissions and limitations
 # under the License.
 """Celery command"""
+from __future__ import annotations
 
 from multiprocessing import Process
-from typing import Optional
 
 import daemon
 import psutil
@@ -83,7 +83,7 @@ def flower(args):
         celery_app.start(options)
 
 
-def _serve_logs(skip_serve_logs: bool = False) -> Optional[Process]:
+def _serve_logs(skip_serve_logs: bool = False) -> Process | None:
     """Starts serve_logs sub-process"""
     if skip_serve_logs is False:
         sub_proc = Process(target=serve_logs)

airflow/cli/commands/cheat_sheet_command.py

@@ -14,7 +14,9 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from typing import Iterable, List, Optional, Union
+from __future__ import annotations
+
+from typing import Iterable
 
 from airflow.cli.cli_parser import ActionCommand, GroupCommand, airflow_commands
 from airflow.cli.simple_table import AirflowConsole, SimpleTable
@@ -31,12 +33,12 @@ def display_commands_index():
     """Display list of all commands."""
 
     def display_recursive(
-        prefix: List[str],
-        commands: Iterable[Union[GroupCommand, ActionCommand]],
-        help_msg: Optional[str] = None,
+        prefix: list[str],
+        commands: Iterable[GroupCommand | ActionCommand],
+        help_msg: str | None = None,
     ):
-        actions: List[ActionCommand] = []
-        groups: List[GroupCommand] = []
+        actions: list[ActionCommand] = []
+        groups: list[GroupCommand] = []
         for command in commands:
             if isinstance(command, GroupCommand):
                 groups.append(command)

airflow/cli/commands/config_command.py

@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Config sub-commands"""
+from __future__ import annotations
+
 import io
 
 import pygments

airflow/cli/commands/connection_command.py

@@ -15,13 +15,15 @@
 # specific language governing permissions and limitations
 # under the License.
 """Connection sub-commands"""
+from __future__ import annotations
+
 import io
 import json
 import os
 import sys
 import warnings
 from pathlib import Path
-from typing import Any, Dict, List
+from typing import Any
 from urllib.parse import urlparse, urlunparse
 
 from sqlalchemy.orm import exc
@@ -38,7 +40,7 @@
 from airflow.utils.session import create_session
 
 
-def _connection_mapper(conn: Connection) -> Dict[str, Any]:
+def _connection_mapper(conn: Connection) -> dict[str, Any]:
     return {
         'id': conn.id,
         'conn_id': conn.conn_id,
@@ -99,7 +101,7 @@ def _connection_to_dict(conn: Connection) -> dict:
     )
 
 
-def _format_connections(conns: List[Connection], file_format: str, serialization_format: str) -> str:
+def _format_connections(conns: list[Connection], file_format: str, serialization_format: str) -> str:
     if serialization_format == 'json':
         serializer_func = lambda x: json.dumps(_connection_to_dict(x))
     elif serialization_format == 'uri':

airflow/cli/commands/dag_command.py

@@ -14,16 +14,16 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
 """Dag sub-commands"""
+from __future__ import annotations
+
 import ast
 import errno
 import json
 import logging
 import signal
 import subprocess
 import sys
-from typing import Optional
 
 from graphviz.dot import Dot
 from sqlalchemy.orm import Session
@@ -295,7 +295,7 @@ def dag_next_execution(args):
             .filter(DagRun.dag_id == dag.dag_id)
             .subquery()
         )
-        max_date_run: Optional[DagRun] = (
+        max_date_run: DagRun | None = (
             session.query(DagRun)
             .filter(DagRun.dag_id == dag.dag_id, DagRun.execution_date == max_date_subq.c.max_date)
             .one_or_none()

airflow/cli/commands/dag_processor_command.py

@@ -14,8 +14,9 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
 """DagProcessor command"""
+from __future__ import annotations
+
 import logging
 from datetime import timedelta
 

airflow/cli/commands/db_command.py

@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Database sub-commands"""
+from __future__ import annotations
+
 import os
 import textwrap
 from tempfile import NamedTemporaryFile

airflow/cli/commands/info_command.py

@@ -15,13 +15,14 @@
 # specific language governing permissions and limitations
 # under the License.
 """Config sub-commands"""
+from __future__ import annotations
+
 import locale
 import logging
 import os
 import platform
 import subprocess
 import sys
-from typing import List, Optional
 from urllib.parse import urlsplit, urlunsplit
 
 import httpx
@@ -132,7 +133,7 @@ class OperatingSystem:
     CYGWIN = "Cygwin"
 
     @staticmethod
-    def get_current() -> Optional[str]:
+    def get_current() -> str | None:
         """Get current operating system"""
         if os.name == "nt":
             return OperatingSystem.WINDOWS
@@ -185,7 +186,7 @@ def __init__(self, anonymizer):
         self.anonymizer = anonymizer
 
     @staticmethod
-    def _get_version(cmd: List[str], grep: Optional[bytes] = None):
+    def _get_version(cmd: list[str], grep: bytes | None = None):
         """Return tools version."""
         try:
             with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) as proc:
@@ -306,7 +307,7 @@ def _paths_info(self):
     def _providers_info(self):
         return [(p.data['package-name'], p.version) for p in ProvidersManager().providers.values()]
 
-    def show(self, output: str, console: Optional[AirflowConsole] = None) -> None:
+    def show(self, output: str, console: AirflowConsole | None = None) -> None:
         """Shows information about Airflow instance"""
         all_info = {
             "Apache Airflow": self._airflow_info,

airflow/cli/commands/jobs_command.py

@@ -14,8 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
-from typing import List
+from __future__ import annotations
 
 from airflow.jobs.base_job import BaseJob
 from airflow.utils.session import provide_session
@@ -39,7 +38,7 @@ def check(args, session=None):
     if args.limit > 0:
         query = query.limit(args.limit)
 
-    jobs: List[BaseJob] = query.all()
+    jobs: list[BaseJob] = query.all()
     alive_jobs = [job for job in jobs if job.is_alive()]
 
     count_alive_jobs = len(alive_jobs)

airflow/cli/commands/kerberos_command.py

@@ -14,8 +14,9 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
 """Kerberos command"""
+from __future__ import annotations
+
 import daemon
 from daemon.pidfile import TimeoutPIDLockFile
 

airflow/cli/commands/kubernetes_command.py

@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Kubernetes sub-commands"""
+from __future__ import annotations
+
 import os
 import sys
 from datetime import datetime, timedelta

airflow/cli/commands/legacy_commands.py

@@ -14,6 +14,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 argparse import ArgumentError
 

airflow/cli/commands/plugins_command.py

@@ -14,8 +14,10 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from __future__ import annotations
+
 import inspect
-from typing import Any, Dict, List, Union
+from typing import Any
 
 from airflow import plugins_manager
 from airflow.cli.simple_table import AirflowConsole
@@ -31,15 +33,15 @@ def _get_name(class_like_object) -> str:
     return class_like_object.__class__.__name__
 
 
-def _join_plugins_names(value: Union[List[Any], Any]) -> str:
+def _join_plugins_names(value: list[Any] | Any) -> str:
     value = value if isinstance(value, list) else [value]
     return ",".join(_get_name(v) for v in value)
 
 
 @suppress_logs_and_warning
 def dump_plugins(args):
     """Dump plugins information"""
-    plugins_info: List[Dict[str, str]] = get_plugin_info()
+    plugins_info: list[dict[str, str]] = get_plugin_info()
     if not plugins_manager.plugins:
         print("No plugins loaded")
         return

airflow/cli/commands/pool_command.py

@@ -16,6 +16,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Pools sub-commands"""
+from __future__ import annotations
+
 import json
 import os
 from json import JSONDecodeError

airflow/cli/commands/provider_command.py

@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Providers sub-commands"""
+from __future__ import annotations
+
 import re
 
 from airflow.cli.simple_table import AirflowConsole

airflow/cli/commands/role_command.py

@@ -15,8 +15,9 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-#
 """Roles sub-commands"""
+from __future__ import annotations
+
 import json
 import os
 

airflow/cli/commands/rotate_fernet_key_command.py

@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Rotate Fernet key command"""
+from __future__ import annotations
+
 from airflow.models import Connection, Variable
 from airflow.utils import cli as cli_utils
 from airflow.utils.session import create_session