diff --git a/airflow/cli/cli_parser.py b/airflow/cli/cli_parser.py
index 21031bd1c21a4..6581cfe632f0d 100644
--- a/airflow/cli/cli_parser.py
+++ b/airflow/cli/cli_parser.py
@@ -17,6 +17,7 @@
 # specific language governing permissions and limitations
 # under the License.
 """Command-line interface"""
+from __future__ import annotations
 
 import argparse
 import json
@@ -24,7 +25,7 @@
 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):
diff --git a/airflow/cli/commands/celery_command.py b/airflow/cli/commands/celery_command.py
index c2deb16ff4c20..4515d8cf959a6 100644
--- a/airflow/cli/commands/celery_command.py
+++ b/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)
diff --git a/airflow/cli/commands/cheat_sheet_command.py b/airflow/cli/commands/cheat_sheet_command.py
index 001a8721330cb..88d9c5940c7ca 100644
--- a/airflow/cli/commands/cheat_sheet_command.py
+++ b/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)
diff --git a/airflow/cli/commands/config_command.py b/airflow/cli/commands/config_command.py
index 1c2674fc811b2..70c194e4afbf8 100644
--- a/airflow/cli/commands/config_command.py
+++ b/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
diff --git a/airflow/cli/commands/connection_command.py b/airflow/cli/commands/connection_command.py
index 8a0c0a3acb80a..3d5bf74113b41 100644
--- a/airflow/cli/commands/connection_command.py
+++ b/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':
diff --git a/airflow/cli/commands/dag_command.py b/airflow/cli/commands/dag_command.py
index 29c4a4f3ff57c..dac0e7b0fc9c8 100644
--- a/airflow/cli/commands/dag_command.py
+++ b/airflow/cli/commands/dag_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.
-
 """Dag sub-commands"""
+from __future__ import annotations
+
 import ast
 import errno
 import json
@@ -23,7 +24,6 @@
 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()
diff --git a/airflow/cli/commands/dag_processor_command.py b/airflow/cli/commands/dag_processor_command.py
index 92f7f37ee16d2..e4493ea536231 100644
--- a/airflow/cli/commands/dag_processor_command.py
+++ b/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
 
diff --git a/airflow/cli/commands/db_command.py b/airflow/cli/commands/db_command.py
index 5f6a84c8a420c..6d981a352063c 100644
--- a/airflow/cli/commands/db_command.py
+++ b/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
diff --git a/airflow/cli/commands/info_command.py b/airflow/cli/commands/info_command.py
index fc03615210af3..f49af3e7dde7d 100644
--- a/airflow/cli/commands/info_command.py
+++ b/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,
diff --git a/airflow/cli/commands/jobs_command.py b/airflow/cli/commands/jobs_command.py
index f6d8d55fb6b9a..76ef0390b9477 100644
--- a/airflow/cli/commands/jobs_command.py
+++ b/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)
diff --git a/airflow/cli/commands/kerberos_command.py b/airflow/cli/commands/kerberos_command.py
index 4ccd07092f474..eec77f6d6f745 100644
--- a/airflow/cli/commands/kerberos_command.py
+++ b/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
 
diff --git a/airflow/cli/commands/kubernetes_command.py b/airflow/cli/commands/kubernetes_command.py
index 7c26821780575..76523b3cb36a6 100644
--- a/airflow/cli/commands/kubernetes_command.py
+++ b/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
diff --git a/airflow/cli/commands/legacy_commands.py b/airflow/cli/commands/legacy_commands.py
index 94f9b690327b2..5643879dc5951 100644
--- a/airflow/cli/commands/legacy_commands.py
+++ b/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
 
diff --git a/airflow/cli/commands/plugins_command.py b/airflow/cli/commands/plugins_command.py
index 2d59e901a187a..2c9f1488c4ff5 100644
--- a/airflow/cli/commands/plugins_command.py
+++ b/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,7 +33,7 @@ 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)
 
@@ -39,7 +41,7 @@ def _join_plugins_names(value: Union[List[Any], Any]) -> str:
 @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
diff --git a/airflow/cli/commands/pool_command.py b/airflow/cli/commands/pool_command.py
index e435c2a4833bc..91e0170775cb7 100644
--- a/airflow/cli/commands/pool_command.py
+++ b/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
diff --git a/airflow/cli/commands/provider_command.py b/airflow/cli/commands/provider_command.py
index 124c8cea8af7b..51a192f61d564 100644
--- a/airflow/cli/commands/provider_command.py
+++ b/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
diff --git a/airflow/cli/commands/role_command.py b/airflow/cli/commands/role_command.py
index 51ba2d27d8373..ed3e4e30b850a 100644
--- a/airflow/cli/commands/role_command.py
+++ b/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
 
diff --git a/airflow/cli/commands/rotate_fernet_key_command.py b/airflow/cli/commands/rotate_fernet_key_command.py
index 9334344d9610b..ca7e956587212 100644
--- a/airflow/cli/commands/rotate_fernet_key_command.py
+++ b/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
diff --git a/airflow/cli/commands/scheduler_command.py b/airflow/cli/commands/scheduler_command.py
index 22a185794fb56..168018887e7ba 100644
--- a/airflow/cli/commands/scheduler_command.py
+++ b/airflow/cli/commands/scheduler_command.py
@@ -14,11 +14,11 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
 """Scheduler command"""
+from __future__ import annotations
+
 import signal
 from multiprocessing import Process
-from typing import Optional
 
 import daemon
 from daemon.pidfile import TimeoutPIDLockFile
@@ -85,7 +85,7 @@ def scheduler(args):
         _run_scheduler_job(args=args)
 
 
-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"""
     from airflow.configuration import conf
     from airflow.utils.serve_logs import serve_logs
@@ -98,7 +98,7 @@ def _serve_logs(skip_serve_logs: bool = False) -> Optional[Process]:
     return None
 
 
-def _serve_health_check(enable_health_check: bool = False) -> Optional[Process]:
+def _serve_health_check(enable_health_check: bool = False) -> Process | None:
     """Starts serve_health_check sub-process"""
     if enable_health_check:
         sub_proc = Process(target=serve_health_check)
diff --git a/airflow/cli/commands/standalone_command.py b/airflow/cli/commands/standalone_command.py
index 3860942adb056..34eae28c23b32 100644
--- a/airflow/cli/commands/standalone_command.py
+++ b/airflow/cli/commands/standalone_command.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
 
 import logging
 import os
@@ -23,7 +24,6 @@
 import threading
 import time
 from collections import deque
-from typing import Dict, List
 
 from termcolor import colored
 
@@ -265,7 +265,7 @@ class SubCommand(threading.Thread):
     complex logic that brings doing line buffering.
     """
 
-    def __init__(self, parent, name: str, command: List[str], env: Dict[str, str]):
+    def __init__(self, parent, name: str, command: list[str], env: dict[str, str]):
         super().__init__()
         self.parent = parent
         self.name = name
diff --git a/airflow/cli/commands/sync_perm_command.py b/airflow/cli/commands/sync_perm_command.py
index d580631fcc8a8..2d3b03133dd04 100644
--- a/airflow/cli/commands/sync_perm_command.py
+++ b/airflow/cli/commands/sync_perm_command.py
@@ -16,6 +16,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Sync permission command"""
+from __future__ import annotations
+
 from airflow.utils import cli as cli_utils
 from airflow.www.app import cached_app
 
diff --git a/airflow/cli/commands/task_command.py b/airflow/cli/commands/task_command.py
index f8916f0466416..9caa8bb4bda7f 100644
--- a/airflow/cli/commands/task_command.py
+++ b/airflow/cli/commands/task_command.py
@@ -16,6 +16,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Task sub-commands"""
+from __future__ import annotations
+
 import datetime
 import importlib
 import json
@@ -23,7 +25,7 @@
 import os
 import textwrap
 from contextlib import contextmanager, redirect_stderr, redirect_stdout, suppress
-from typing import Dict, Generator, List, Optional, Tuple, Union
+from typing import Generator, Union
 
 from pendulum.parsing.exceptions import ParserError
 from sqlalchemy.orm.exc import NoResultFound
@@ -76,9 +78,9 @@ def _get_dag_run(
     *,
     dag: DAG,
     create_if_necessary: CreateIfNecessary,
-    exec_date_or_run_id: Optional[str] = None,
+    exec_date_or_run_id: str | None = None,
     session: Session,
-) -> Tuple[DagRun, bool]:
+) -> tuple[DagRun, bool]:
     """Try to retrieve a DAG run from a string representing either a run ID or logical date.
 
     This checks DAG runs like this:
@@ -94,7 +96,7 @@ def _get_dag_run(
     """
     if not exec_date_or_run_id and not create_if_necessary:
         raise ValueError("Must provide `exec_date_or_run_id` if not `create_if_necessary`.")
-    execution_date: Optional[datetime.datetime] = None
+    execution_date: datetime.datetime | None = None
     if exec_date_or_run_id:
         dag_run = dag.get_dagrun(run_id=exec_date_or_run_id, session=session)
         if dag_run:
@@ -140,11 +142,11 @@ def _get_ti(
     task: BaseOperator,
     map_index: int,
     *,
-    exec_date_or_run_id: Optional[str] = None,
-    pool: Optional[str] = None,
+    exec_date_or_run_id: str | None = None,
+    pool: str | None = None,
     create_if_necessary: CreateIfNecessary = False,
     session: Session = NEW_SESSION,
-) -> Tuple[TaskInstance, bool]:
+) -> tuple[TaskInstance, bool]:
     """Get the task instance through DagRun.run_id, if that fails, get the TI the old way"""
     if not exec_date_or_run_id and not create_if_necessary:
         raise ValueError("Must provide `exec_date_or_run_id` if not `create_if_necessary`.")
@@ -267,7 +269,7 @@ def _run_raw_task(args, ti: TaskInstance) -> None:
     )
 
 
-def _extract_external_executor_id(args) -> Optional[str]:
+def _extract_external_executor_id(args) -> str | None:
     if hasattr(args, "external_executor_id"):
         return getattr(args, "external_executor_id")
     return os.environ.get("external_executor_id", None)
@@ -441,7 +443,7 @@ def task_list(args, dag=None):
         print("\n".join(tasks))
 
 
-SUPPORTED_DEBUGGER_MODULES: List[str] = [
+SUPPORTED_DEBUGGER_MODULES: list[str] = [
     "pudb",
     "web_pdb",
     "ipdb",
@@ -498,7 +500,7 @@ def task_states_for_dag_run(args, session=None):
 
     has_mapped_instances = any(ti.map_index >= 0 for ti in dag_run.task_instances)
 
-    def format_task_instance(ti: TaskInstance) -> Dict[str, str]:
+    def format_task_instance(ti: TaskInstance) -> dict[str, str]:
         data = {
             "dag_id": ti.dag_id,
             "execution_date": dag_run.execution_date.isoformat(),
diff --git a/airflow/cli/commands/triggerer_command.py b/airflow/cli/commands/triggerer_command.py
index 70fbbda2fcbb3..f364f708c4509 100644
--- a/airflow/cli/commands/triggerer_command.py
+++ b/airflow/cli/commands/triggerer_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.
-
 """Triggerer command"""
+from __future__ import annotations
+
 import signal
 
 import daemon
diff --git a/airflow/cli/commands/user_command.py b/airflow/cli/commands/user_command.py
index ddbb7cfc82d6c..75a6899662dea 100644
--- a/airflow/cli/commands/user_command.py
+++ b/airflow/cli/commands/user_command.py
@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """User sub-commands"""
+from __future__ import annotations
+
 import functools
 import getpass
 import json
@@ -22,7 +24,7 @@
 import random
 import re
 import string
-from typing import Any, Dict, List
+from typing import Any
 
 from marshmallow import Schema, fields, validate
 from marshmallow.exceptions import ValidationError
@@ -189,7 +191,7 @@ def users_import(args):
         print("Updated the following users:\n\t{}".format("\n\t".join(users_updated)))
 
 
-def _import_users(users_list: List[Dict[str, Any]]):
+def _import_users(users_list: list[dict[str, Any]]):
     appbuilder = cached_app().appbuilder
     users_created = []
     users_updated = []
diff --git a/airflow/cli/commands/variable_command.py b/airflow/cli/commands/variable_command.py
index 40eb193148a80..426972328134b 100644
--- a/airflow/cli/commands/variable_command.py
+++ b/airflow/cli/commands/variable_command.py
@@ -16,6 +16,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Variable subcommands"""
+from __future__ import annotations
+
 import json
 import os
 from json import JSONDecodeError
diff --git a/airflow/cli/commands/version_command.py b/airflow/cli/commands/version_command.py
index 7e5190185884f..365e9b2316ad3 100644
--- a/airflow/cli/commands/version_command.py
+++ b/airflow/cli/commands/version_command.py
@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 """Version command"""
+from __future__ import annotations
+
 import airflow
 
 
diff --git a/airflow/cli/commands/webserver_command.py b/airflow/cli/commands/webserver_command.py
index d93d44a2b79fc..9d15d44b7dfcd 100644
--- a/airflow/cli/commands/webserver_command.py
+++ b/airflow/cli/commands/webserver_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.
-
 """Webserver command"""
+from __future__ import annotations
+
 import hashlib
 import logging
 import os
@@ -26,7 +27,7 @@
 import time
 from contextlib import suppress
 from time import sleep
-from typing import Dict, List, NoReturn
+from typing import NoReturn
 
 import daemon
 import psutil
@@ -102,7 +103,7 @@ def __init__(
         self._last_plugin_state = self._generate_plugin_state() if reload_on_plugin_change else None
         self._restart_on_next_plugin_check = False
 
-    def _generate_plugin_state(self) -> Dict[str, float]:
+    def _generate_plugin_state(self) -> dict[str, float]:
         """
         Generate dict of filenames and last modification time of all files in settings.PLUGINS_FOLDER
         directory.
@@ -110,7 +111,7 @@ def _generate_plugin_state(self) -> Dict[str, float]:
         if not settings.PLUGINS_FOLDER:
             return {}
 
-        all_filenames: List[str] = []
+        all_filenames: list[str] = []
         for (root, _, filenames) in os.walk(settings.PLUGINS_FOLDER):
             all_filenames.extend(os.path.join(root, f) for f in filenames)
         plugin_state = {f: self._get_file_hash(f) for f in sorted(all_filenames)}
diff --git a/airflow/cli/simple_table.py b/airflow/cli/simple_table.py
index efd418d21fcbe..7c372ea1f241f 100644
--- a/airflow/cli/simple_table.py
+++ b/airflow/cli/simple_table.py
@@ -14,9 +14,11 @@
 # 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
 import json
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Callable
 
 from rich.box import ASCII_DOUBLE_HEAD
 from rich.console import Console
@@ -40,17 +42,17 @@ def __init__(self, show_header: bool = True, *args, **kwargs):
         # If show header in tables
         self.show_header = show_header
 
-    def print_as_json(self, data: Dict):
+    def print_as_json(self, data: dict):
         """Renders dict as json text representation"""
         json_content = json.dumps(data)
         self.print(Syntax(json_content, "json", theme="ansi_dark"), soft_wrap=True)
 
-    def print_as_yaml(self, data: Dict):
+    def print_as_yaml(self, data: dict):
         """Renders dict as yaml text representation"""
         yaml_content = yaml.dump(data)
         self.print(Syntax(yaml_content, "yaml", theme="ansi_dark"), soft_wrap=True)
 
-    def print_as_table(self, data: List[Dict]):
+    def print_as_table(self, data: list[dict]):
         """Renders list of dictionaries as table"""
         if not data:
             self.print("No data found")
@@ -64,7 +66,7 @@ def print_as_table(self, data: List[Dict]):
             table.add_row(*(str(d) for d in row.values()))
         self.print(table)
 
-    def print_as_plain_table(self, data: List[Dict]):
+    def print_as_plain_table(self, data: list[dict]):
         """Renders list of dictionaries as a simple table than can be easily piped"""
         if not data:
             self.print("No data found")
@@ -73,7 +75,7 @@ def print_as_plain_table(self, data: List[Dict]):
         output = tabulate(rows, tablefmt="plain", headers=list(data[0].keys()))
         print(output)
 
-    def _normalize_data(self, value: Any, output: str) -> Optional[Union[list, str, dict]]:
+    def _normalize_data(self, value: Any, output: str) -> list | str | dict | None:
         if isinstance(value, (tuple, list)):
             if output == "table":
                 return ",".join(str(self._normalize_data(x, output)) for x in value)
@@ -86,9 +88,9 @@ def _normalize_data(self, value: Any, output: str) -> Optional[Union[list, str,
             return None
         return str(value)
 
-    def print_as(self, data: List[Union[Dict, Any]], output: str, mapper: Optional[Callable] = None):
+    def print_as(self, data: list[dict | Any], output: str, mapper: Callable | None = None):
         """Prints provided using format specified by output argument"""
-        output_to_renderer: Dict[str, Callable[[Any], None]] = {
+        output_to_renderer: dict[str, Callable[[Any], None]] = {
             "json": self.print_as_json,
             "yaml": self.print_as_yaml,
             "table": self.print_as_table,
@@ -104,7 +106,7 @@ def print_as(self, data: List[Union[Dict, Any]], output: str, mapper: Optional[C
             raise ValueError("To tabulate non-dictionary data you need to provide `mapper` function")
 
         if mapper:
-            dict_data: List[Dict] = [mapper(d) for d in data]
+            dict_data: list[dict] = [mapper(d) for d in data]
         else:
             dict_data = data
         dict_data = [{k: self._normalize_data(v, output) for k, v in d.items()} for d in dict_data]