Revert "Upgrade mypy to 0.991 (#28926)" (#29470)

This reverts commit 6ae0a80.

.pre-commit-config.yaml

@@ -160,7 +160,7 @@ repos:
         entry: ./scripts/ci/pre_commit/pre_commit_update_common_sql_api_stubs.py
         language: python
         files: ^scripts/ci/pre_commit/pre_commit_update_common_sql_api\.py|^airflow/providers/common/sql/.*\.pyi?$
-        additional_dependencies: ['rich>=12.4.4', 'mypy==0.991', 'black==22.12.0', 'jinja2']
+        additional_dependencies: ['rich>=12.4.4', 'mypy==0.971', 'black==22.12.0', 'jinja2']
         pass_filenames: false
         require_serial: true
       - id: update-black-version

airflow/dag_processing/manager.py

@@ -141,7 +141,8 @@ def __init__(
 
     def start(self) -> None:
         """Launch DagFileProcessorManager processor and start DAG parsing loop in manager."""
-        context = self._get_multiprocessing_context()
+        mp_start_method = self._get_multiprocessing_start_method()
+        context = multiprocessing.get_context(mp_start_method)
         self._last_parsing_stat_received_at = time.monotonic()
 
         self._parent_signal_conn, child_signal_conn = context.Pipe()

airflow/dag_processing/processor.py

@@ -187,7 +187,8 @@ def _handle_dag_file_processing():
 
     def start(self) -> None:
         """Launch the process and start processing the DAG."""
-        context = self._get_multiprocessing_context()
+        start_method = self._get_multiprocessing_start_method()
+        context = multiprocessing.get_context(start_method)
 
         _parent_channel, _child_channel = context.Pipe(duplex=False)
         process = context.Process(

airflow/decorators/base.py

@@ -489,7 +489,7 @@ def _get_unmap_kwargs(self, mapped_kwargs: Mapping[str, Any], *, strict: bool) -
         return super()._get_unmap_kwargs(kwargs, strict=False)
 
 
-class Task(Protocol, Generic[FParams, FReturn]):
+class Task(Generic[FParams, FReturn]):
     """Declaration of a @task-decorated callable for type-checking.
 
     An instance of this type inherits the call signature of the decorated

airflow/example_dags/plugins/workday.py

@@ -78,9 +78,6 @@ def next_dagrun_info(
 
     # [END howto_timetable_next_dagrun_info]
 
-    def validate(self):
-        pass
-
 
 class WorkdayTimetablePlugin(AirflowPlugin):
     name = "workday_timetable_plugin"

airflow/executors/kubernetes_executor.py

@@ -69,7 +69,7 @@
 class ResourceVersion:
     """Singleton for tracking resourceVersion from Kubernetes."""
 
-    _instance: ResourceVersion | None = None
+    _instance = None
     resource_version: dict[str, str] = {}
 
     def __new__(cls):

airflow/hooks/base.py

@@ -90,11 +90,11 @@ def get_conn(self) -> Any:
 
     @classmethod
     def get_connection_form_widgets(cls) -> dict[str, Any]:
-        return {}
+        ...
 
     @classmethod
     def get_ui_field_behaviour(cls) -> dict[str, Any]:
-        return {}
+        ...
 
 
 class DiscoverableHook(Protocol):

airflow/listeners/listener.py

@@ -30,7 +30,7 @@
 log = logging.getLogger(__name__)
 
 
-_listener_manager: ListenerManager | None = None
+_listener_manager = None
 
 
 class ListenerManager:

airflow/models/baseoperator.py

@@ -30,7 +30,7 @@
 from abc import ABCMeta, abstractmethod
 from datetime import datetime, timedelta
 from inspect import signature
-from types import ClassMethodDescriptorType, FunctionType
+from types import FunctionType
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -169,7 +169,7 @@ def get_merged_defaults(
 class _PartialDescriptor:
     """A descriptor that guards against ``.partial`` being called on Task objects."""
 
-    class_method: ClassMethodDescriptorType | None = None
+    class_method = None
 
     def __get__(
         self, obj: BaseOperator, cls: type[BaseOperator] | None = None

airflow/providers/google/cloud/hooks/cloud_sql.py

@@ -691,6 +691,8 @@ class CloudSQLDatabaseHook(BaseHook):
     conn_type = "gcpcloudsqldb"
     hook_name = "Google Cloud SQL Database"
 
+    _conn = None
+
     def __init__(
         self,
         gcp_cloudsql_conn_id: str = "google_cloud_sql_default",

airflow/providers/google/cloud/hooks/functions.py

@@ -40,7 +40,7 @@ class CloudFunctionsHook(GoogleBaseHook):
     keyword arguments rather than positional.
     """
 
-    _conn: build | None = None
+    _conn = None
 
     def __init__(
         self,

airflow/providers/google/cloud/hooks/gcs.py

@@ -51,18 +51,8 @@
 from airflow.utils import timezone
 from airflow.version import version
 
-try:
-    # Airflow 2.3 doesn't have this yet
-    from airflow.typing_compat import ParamSpec
-except ImportError:
-    try:
-        from typing import ParamSpec  # type: ignore[no-redef,attr-defined]
-    except ImportError:
-        from typing_extensions import ParamSpec
-
 RT = TypeVar("RT")
 T = TypeVar("T", bound=Callable)
-FParams = ParamSpec("FParams")
 
 # GCSHook has a method named 'list' (to junior devs: please don't do this), so
 # we need to create an alias to prevent Mypy being confused.
@@ -86,7 +76,7 @@ def _fallback_object_url_to_object_name_and_bucket_name(
     :return: Decorator
     """
 
-    def _wrapper(func: Callable[FParams, RT]) -> Callable[FParams, RT]:
+    def _wrapper(func: T):
         @functools.wraps(func)
         def _inner_wrapper(self: GCSHook, *args, **kwargs) -> RT:
             if args:
@@ -127,11 +117,11 @@ def _inner_wrapper(self: GCSHook, *args, **kwargs) -> RT:
                     f"'{bucket_name_keyword_arg_name}'"
                 )
 
-            return func(self, *args, **kwargs)  # type: ignore
+            return func(self, *args, **kwargs)
 
-        return cast(Callable[FParams, RT], _inner_wrapper)
+        return cast(T, _inner_wrapper)
 
-    return cast(Callable[[T], T], _wrapper)
+    return _wrapper
 
 
 # A fake bucket to use in functions decorated by _fallback_object_url_to_object_name_and_bucket_name.

airflow/providers/google/cloud/hooks/life_sciences.py

@@ -54,7 +54,7 @@ class LifeSciencesHook(GoogleBaseHook):
         account from the list granting this role to the originating account.
     """
 
-    _conn: build | None = None
+    _conn = None
 
     def __init__(
         self,

airflow/providers/google/cloud/utils/field_validator.py

@@ -206,7 +206,7 @@ def _sanity_checks(
         full_field_path: str,
         regexp: str,
         allow_empty: bool,
-        custom_validation: Callable | None,
+        custom_validation: Callable,
         value,
     ) -> None:
         if value is None and field_type != "union":

airflow/providers/google/firebase/hooks/firestore.py

@@ -53,7 +53,7 @@ class CloudFirestoreHook(GoogleBaseHook):
         account from the list granting this role to the originating account.
     """
 
-    _conn: build | None = None
+    _conn = None
 
     def __init__(
         self,

airflow/providers/google/marketing_platform/hooks/search_ads.py

@@ -28,7 +28,7 @@
 class GoogleSearchAdsHook(GoogleBaseHook):
     """Hook for Google Search Ads 360."""
 
-    _conn: build | None = None
+    _conn = None
 
     def __init__(
         self,

airflow/providers/microsoft/psrp/hooks/psrp.py

@@ -75,7 +75,8 @@ class PsrpHook(BaseHook):
     or by setting this key as the extra fields of your connection.
     """
 
-    _conn: RunspacePool | None = None
+    _conn = None
+    _configuration_name = None
     _wsman_ref: WeakKeyDictionary[RunspacePool, WSMan] = WeakKeyDictionary()
 
     def __init__(

airflow/settings.py

@@ -411,7 +411,7 @@ def dispose_orm():
     global engine
     global Session
 
-    if Session is not None:
+    if Session:
         Session.remove()
         Session = None
     if engine:

airflow/stats.py

@@ -74,10 +74,9 @@ def timing(cls, stat: str, dt: float | datetime.timedelta, tags: dict[str, str]
     @classmethod
     def timer(cls, *args, **kwargs) -> TimerProtocol:
         """Timer metric that can be cancelled."""
-        raise NotImplementedError()
 
 
-class Timer(TimerProtocol):
+class Timer:
     """
     Timer that records duration, and optional sends to StatsD backend.
 
@@ -361,7 +360,7 @@ def timer(self, stat=None, *args, tags=None, **kwargs):
 
 
 class _Stats(type):
-    factory: Callable[[], StatsLogger]
+    factory = None
     instance: StatsLogger | None = None
 
     def __getattr__(cls, name):
@@ -375,7 +374,7 @@ def __getattr__(cls, name):
 
     def __init__(cls, *args, **kwargs):
         super().__init__(cls)
-        if not hasattr(cls.__class__, "factory"):
+        if cls.__class__.factory is None:
             is_datadog_enabled_defined = conf.has_option("metrics", "statsd_datadog_enabled")
             if is_datadog_enabled_defined and conf.getboolean("metrics", "statsd_datadog_enabled"):
                 cls.__class__.factory = cls.get_dogstatsd_logger

airflow/timetables/simple.py

@@ -55,9 +55,6 @@ def serialize(self) -> dict[str, Any]:
     def infer_manual_data_interval(self, *, run_after: DateTime) -> DataInterval:
         return DataInterval.exact(run_after)
 
-    def validate(self):
-        pass
-
 
 class NullTimetable(_TrivialTimetable):
     """Timetable that never schedules anything.

airflow/utils/mixins.py

@@ -25,8 +25,6 @@
 from airflow.utils.context import Context
 
 if typing.TYPE_CHECKING:
-    import multiprocessing.context
-
     from airflow.models.operator import Operator
 
 
@@ -46,10 +44,6 @@ def _get_multiprocessing_start_method(self) -> str:
             raise ValueError("Failed to determine start method")
         return method
 
-    def _get_multiprocessing_context(self) -> multiprocessing.context.DefaultContext:
-        mp_start_method = self._get_multiprocessing_start_method()
-        return multiprocessing.get_context(mp_start_method)  # type: ignore
-
 
 class ResolveMixin:
     """A runtime-resolved value."""

airflow/utils/session.py

@@ -28,7 +28,7 @@
 @contextlib.contextmanager
 def create_session() -> Generator[settings.SASession, None, None]:
     """Contextmanager that will create and teardown a session."""
-    if not hasattr(settings, "Session") or settings.Session is None:
+    if not settings.Session:
         raise RuntimeError("Session must be set before!")
     session = settings.Session()
     try:

airflow/utils/state.py

@@ -85,7 +85,7 @@ class State:
     FAILED = TaskInstanceState.FAILED
 
     # These are TaskState only
-    NONE: None = None
+    NONE = None
     REMOVED = TaskInstanceState.REMOVED
     SCHEDULED = TaskInstanceState.SCHEDULED
     QUEUED = TaskInstanceState.QUEUED

airflow/www/extensions/init_appbuilder.py

@@ -14,8 +14,6 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
-# mypy: disable-error-code=var-annotated
 from __future__ import annotations
 
 import logging

airflow/www/fab_security/manager.py

@@ -14,8 +14,6 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
-# mypy: disable-error-code=var-annotated
 from __future__ import annotations
 
 import base64

dev/breeze/src/airflow_breeze/utils/click_utils.py

@@ -19,4 +19,4 @@
 try:
     from rich_click import RichGroup as BreezeGroup
 except ImportError:
-    from click import Group as BreezeGroup  # type: ignore[assignment] # noqa
+    from click import Group as BreezeGroup  # type: ignore[misc] # noqa

scripts/ci/pre_commit/pre_commit_check_order_dockerfile_extras.py

@@ -28,7 +28,7 @@
 
 from rich import print
 
-errors: list[str] = []
+errors = []
 
 MY_DIR_PATH = Path(__file__).parent.resolve()
 

scripts/ci/pre_commit/pre_commit_check_order_setup.py

@@ -30,7 +30,7 @@
 
 from rich import print
 
-errors: list[str] = []
+errors = []
 
 MY_DIR_PATH = os.path.dirname(__file__)
 SOURCE_DIR_PATH = os.path.abspath(os.path.join(MY_DIR_PATH, os.pardir, os.pardir, os.pardir))

scripts/ci/pre_commit/pre_commit_check_system_tests_hidden_in_index.py

@@ -19,7 +19,6 @@
 
 import sys
 from pathlib import Path
-from typing import Any
 
 from rich.console import Console
 
@@ -38,7 +37,7 @@
 PREFIX = "apache-airflow-providers-"
 
 
-errors: list[Any] = []
+errors = []
 
 
 def check_system_test_entry_hidden(provider_index: Path):