Use Literal from airflow.typing_compat in Airflow core (apache#33821)

airflow/cli/commands/task_command.py

@@ -31,7 +31,6 @@
 from pendulum.parsing.exceptions import ParserError
 from sqlalchemy import select
 from sqlalchemy.orm.exc import NoResultFound
-from typing_extensions import Literal
 
 from airflow import settings
 from airflow.cli.simple_table import AirflowConsole
@@ -50,6 +49,7 @@
 from airflow.settings import IS_K8S_EXECUTOR_POD
 from airflow.ti_deps.dep_context import DepContext
 from airflow.ti_deps.dependencies_deps import SCHEDULER_QUEUED_DEPS
+from airflow.typing_compat import Literal
 from airflow.utils import cli as cli_utils
 from airflow.utils.cli import (
     get_dag,

airflow/models/dag.py

@@ -131,13 +131,13 @@
     from pendulum.tz.timezone import Timezone
     from sqlalchemy.orm.query import Query
     from sqlalchemy.orm.session import Session
-    from typing_extensions import Literal
 
     from airflow.datasets import Dataset
     from airflow.decorators import TaskDecoratorCollection
     from airflow.models.dagbag import DagBag
     from airflow.models.operator import Operator
     from airflow.models.slamiss import SlaMiss
+    from airflow.typing_compat import Literal
     from airflow.utils.task_group import TaskGroup
 
 log = logging.getLogger(__name__)

airflow/models/dagrun.py

@@ -72,10 +72,10 @@
     from datetime import datetime
 
     from sqlalchemy.orm import Query, Session
-    from typing_extensions import Literal
 
     from airflow.models.dag import DAG
     from airflow.models.operator import Operator
+    from airflow.typing_compat import Literal
     from airflow.utils.types import ArgNotSet
 
     CreatedTasks = TypeVar("CreatedTasks", Iterator["dict[str, Any]"], Iterator[TI])

airflow/models/mappedoperator.py

@@ -25,7 +25,6 @@
 from typing import TYPE_CHECKING, Any, ClassVar, Collection, Iterable, Iterator, Mapping, Sequence, Union
 
 import attr
-from typing_extensions import Literal
 
 from airflow import settings
 from airflow.compat.functools import cache
@@ -52,6 +51,7 @@
 from airflow.models.pool import Pool
 from airflow.serialization.enums import DagAttributeTypes
 from airflow.ti_deps.deps.mapped_task_expanded import MappedTaskIsExpanded
+from airflow.typing_compat import Literal
 from airflow.utils.context import context_update_for_unmapped
 from airflow.utils.helpers import is_container, prevent_duplicates
 from airflow.utils.types import NOTSET

airflow/models/taskinstance.py

@@ -135,7 +135,6 @@
     from sqlalchemy.orm.session import Session
     from sqlalchemy.sql.elements import BooleanClauseList
     from sqlalchemy.sql.expression import ColumnOperators
-    from typing_extensions import Literal
 
     from airflow.models.abstractoperator import TaskStateChangeCallback
     from airflow.models.baseoperator import BaseOperator
@@ -144,7 +143,7 @@
     from airflow.models.dataset import DatasetEvent
     from airflow.models.operator import Operator
     from airflow.timetables.base import DataInterval
-    from airflow.typing_compat import TypeGuard
+    from airflow.typing_compat import Literal, TypeGuard
     from airflow.utils.task_group import TaskGroup
 
     # This is a workaround because mypy doesn't work with hybrid_property

airflow/providers_manager.py

@@ -37,6 +37,7 @@
 from airflow.exceptions import AirflowOptionalProviderFeatureException
 from airflow.hooks.filesystem import FSHook
 from airflow.hooks.package_index import PackageIndexHook
+from airflow.typing_compat import Literal
 from airflow.utils import yaml
 from airflow.utils.entry_points import entry_points_with_dist
 from airflow.utils.log.logging_mixin import LoggingMixin

airflow/typing_compat.py

@@ -31,8 +31,14 @@
 from typing import Protocol, TypedDict, runtime_checkable
 
 # Literal from typing module has various issues in different Python versions, see:
-# https://typing-extensions.readthedocs.io/en/latest/#Literal
-from typing_extensions import Literal
+# - https://typing-extensions.readthedocs.io/en/latest/#Literal
+# - bpo-45679: https://github.com/python/cpython/pull/29334
+# - bpo-42345: https://github.com/python/cpython/pull/23294
+# - bpo-42345: https://github.com/python/cpython/pull/23383
+if sys.version_info >= (3, 10, 1) or (3, 9, 8) <= sys.version_info < (3, 10):
+    from typing import Literal
+else:
+    from typing_extensions import Literal  # type: ignore[assignment]
 
 if sys.version_info >= (3, 10):
     from typing import ParamSpec, TypeGuard

airflow/utils/dates.py

@@ -23,9 +23,9 @@
 
 from croniter import croniter
 from dateutil.relativedelta import relativedelta  # for doctest
-from typing_extensions import Literal
 
 from airflow.exceptions import RemovedInAirflow3Warning
+from airflow.typing_compat import Literal
 from airflow.utils import timezone
 
 cron_presets: dict[str, str] = {