diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 03ea7c8353e88..e73026918e778 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -429,7 +429,6 @@ jobs:
     needs: [build-info, ci-images]
     env:
       RUNS_ON: ${{ fromJson(needs.build-info.outputs.runsOn) }}
-      SKIP: "identity"
       MOUNT_SELECTED_LOCAL_SOURCES: "true"
       PYTHON_MAJOR_MINOR_VERSION: ${{needs.build-info.outputs.defaultPythonVersion}}
     if: needs.build-info.outputs.basic-checks-only == 'false'
@@ -471,10 +470,17 @@ ${{ hashFiles('.pre-commit-config.yaml') }}"
         with:
           path: 'airflow/ui/node_modules'
           key: ${{ runner.os }}-ui-node-modules-${{ hashFiles('airflow/ui/**/yarn.lock') }}
-      - name: "Static checks"
+      - name: "Static checks (no mypy)"
         run: ./scripts/ci/static_checks/run_static_checks.sh
         env:
           VERBOSE: false
+          SKIP: "identity,mypy"
+      - name: "MyPy static checks"
+        run: ./scripts/ci/static_checks/run_static_checks.sh mypy
+        env:
+          VERBOSE: false
+          DO_NOT_FAIL_ON_ERROR: "true"
+          COLUMNS: 300
 
   # Those checks are run if no image needs to be built for checks. This is for simple changes that
   # Do not touch any of the python code or any of the important files that might require building
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index e06cc4eaa7374..81931601886ca 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -721,22 +721,23 @@ repos:
       - id: mypy
         name: Run mypy
         language: system
-        entry: ./scripts/ci/pre_commit/pre_commit_mypy.sh
+        entry: ./scripts/ci/pre_commit/pre_commit_mypy.sh --namespace-packages
         files: \.py$
-        exclude: ^dev|^provider_packages|^chart|^docs|^airflow/_vendor/
+        exclude: ^provider_packages|^chart|^docs|^airflow/_vendor/
+        require_serial: true
       - id: mypy
         name: Run mypy for helm chart tests
         language: system
         entry: ./scripts/ci/pre_commit/pre_commit_mypy.sh
         files: ^chart/.*\.py$
-        require_serial: false
+        require_serial: true
       - id: mypy
         name: Run mypy for /docs/ folder
         language: system
         entry: ./scripts/ci/pre_commit/pre_commit_mypy.sh
         files: ^docs/.*\.py$
         exclude: rtd-deprecation
-        require_serial: false
+        require_serial: true
       - id: flake8
         name: Run flake8
         language: system
diff --git a/airflow/api_connexion/endpoints/dag_endpoint.py b/airflow/api_connexion/endpoints/dag_endpoint.py
index df68ee10d6c3a..415e414e3460a 100644
--- a/airflow/api_connexion/endpoints/dag_endpoint.py
+++ b/airflow/api_connexion/endpoints/dag_endpoint.py
@@ -32,7 +32,6 @@
 from airflow.exceptions import AirflowException, DagNotFound
 from airflow.models.dag import DagModel, DagTag
 from airflow.security import permissions
-from airflow.settings import Session
 from airflow.utils.session import provide_session
 
 
@@ -111,15 +110,15 @@ def patch_dag(session, dag_id, update_mask=None):
 
 @security.requires_access([(permissions.ACTION_CAN_DELETE, permissions.RESOURCE_DAG)])
 @provide_session
-def delete_dag(dag_id: str, session: Session):
+def delete_dag(dag_id: str, session):
     """Delete the specific DAG."""
     # TODO: This function is shared with the /delete endpoint used by the web
     # UI, so we're reusing it to simplify maintenance. Refactor the function to
     # another place when the experimental/legacy API is removed.
-    from airflow.api.common.experimental import delete_dag
+    from airflow.api.common.experimental import delete_dag as delete_dag_module
 
     try:
-        delete_dag.delete_dag(dag_id, session=session)
+        delete_dag_module.delete_dag(dag_id, session=session)
     except DagNotFound:
         raise NotFound(f"Dag with id: '{dag_id}' not found")
     except AirflowException:
diff --git a/airflow/api_connexion/schemas/dag_schema.py b/airflow/api_connexion/schemas/dag_schema.py
index 8e3ed2096a859..c5c9034fff4a4 100644
--- a/airflow/api_connexion/schemas/dag_schema.py
+++ b/airflow/api_connexion/schemas/dag_schema.py
@@ -86,7 +86,7 @@ class DAGDetailSchema(DAGSchema):
     doc_md = fields.String()
     default_view = fields.String()
     params = fields.Method('get_params', dump_only=True)
-    tags = fields.Method("get_tags", dump_only=True)
+    tags = fields.Method("get_tags", dump_only=True)  # type: ignore
     is_paused = fields.Method("get_is_paused", dump_only=True)
     is_active = fields.Method("get_is_active", dump_only=True)
 
diff --git a/airflow/cli/commands/cheat_sheet_command.py b/airflow/cli/commands/cheat_sheet_command.py
index 0dcce3c84f044..72fcff5fc20c5 100644
--- a/airflow/cli/commands/cheat_sheet_command.py
+++ b/airflow/cli/commands/cheat_sheet_command.py
@@ -39,7 +39,7 @@ def display_recursive(
         actions: List[ActionCommand]
         groups: List[GroupCommand]
         actions_iter, groups_iter = partition(lambda x: isinstance(x, GroupCommand), commands)
-        actions, groups = list(actions_iter), list(groups_iter)
+        actions, groups = list(actions_iter), list(groups_iter)  # type: ignore
 
         console = AirflowConsole()
         if actions:
diff --git a/airflow/cli/commands/info_command.py b/airflow/cli/commands/info_command.py
index 590a96ce4911b..5a6ccdf39b7d9 100644
--- a/airflow/cli/commands/info_command.py
+++ b/airflow/cli/commands/info_command.py
@@ -57,7 +57,7 @@ class NullAnonymizer(Anonymizer):
     def _identity(self, value):
         return value
 
-    process_path = process_username = process_url = _identity
+    process_path = process_username = process_url = _identity  # type: ignore
 
     del _identity
 
diff --git a/airflow/cli/commands/role_command.py b/airflow/cli/commands/role_command.py
index a3098e9d0afe6..4e403dc75dd4f 100644
--- a/airflow/cli/commands/role_command.py
+++ b/airflow/cli/commands/role_command.py
@@ -24,7 +24,7 @@
 from airflow.utils import cli as cli_utils
 from airflow.utils.cli import suppress_logs_and_warning
 from airflow.www.app import cached_app
-from airflow.www.security import EXISTING_ROLES
+from airflow.www.security import EXISTING_ROLES  # type: ignore
 
 
 @suppress_logs_and_warning
diff --git a/airflow/cli/simple_table.py b/airflow/cli/simple_table.py
index 30f29e3249733..a4bc1abd16bd9 100644
--- a/airflow/cli/simple_table.py
+++ b/airflow/cli/simple_table.py
@@ -70,13 +70,13 @@ def print_as_plain_table(self, data: List[Dict]):
             self.print("No data found")
             return
         rows = [d.values() for d in data]
-        output = tabulate(rows, tablefmt="plain", headers=data[0].keys())
+        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]]:
         if isinstance(value, (tuple, list)):
             if output == "table":
-                return ",".join(self._normalize_data(x, output) for x in value)
+                return ",".join(str(self._normalize_data(x, output)) for x in value)
             return [self._normalize_data(x, output) for x in value]
         if isinstance(value, dict) and output != "table":
             return {k: self._normalize_data(v, output) for k, v in value.items()}
@@ -106,9 +106,9 @@ def print_as(self, data: List[Union[Dict, Any]], output: str, mapper: Optional[C
         if mapper:
             dict_data: List[Dict] = [mapper(d) for d in data]
         else:
-            dict_data: List[Dict] = data
+            dict_data = data
         dict_data = [{k: self._normalize_data(v, output) for k, v in d.items()} for d in dict_data]
-        renderer(dict_data)
+        renderer(dict_data)  # type: ignore
 
 
 class SimpleTable(Table):
diff --git a/airflow/compat/asyncio.py b/airflow/compat/asyncio.py
index 7e9525ca76f53..6e0dd06c0987d 100644
--- a/airflow/compat/asyncio.py
+++ b/airflow/compat/asyncio.py
@@ -20,6 +20,6 @@
 except ImportError:
     # create_task is not present in Python 3.6. Once Airflow is at 3.7+, we can
     # remove this helper.
-    def create_task(*args, **kwargs):
+    def create_task(*args, **kwargs):  # type: ignore
         """A version of create_task that always errors."""
         raise RuntimeError("Airflow's async functionality is only available on Python 3.7+")
diff --git a/airflow/config_templates/default_webserver_config.py b/airflow/config_templates/default_webserver_config.py
index bc0bb2fbdb74b..5180a42b7f3e0 100644
--- a/airflow/config_templates/default_webserver_config.py
+++ b/airflow/config_templates/default_webserver_config.py
@@ -18,12 +18,12 @@
 """Default configuration for the Airflow webserver"""
 import os
 
-from airflow.www.fab_security.manager import AUTH_DB
+from airflow.www.fab_security.manager import AUTH_DB  # type: ignore
 
-# from airflow.www.fab_security.manager import AUTH_LDAP
-# from airflow.www.fab_security.manager import AUTH_OAUTH
-# from airflow.www.fab_security.manager import AUTH_OID
-# from airflow.www.fab_security.manager import AUTH_REMOTE_USER
+# from airflow.www.fab_security.manager import AUTH_LDAP  # type: ignore
+# from airflow.www.fab_security.manager import AUTH_OAUTH  # type: ignore
+# from airflow.www.fab_security.manager import AUTH_OID  # type: ignore
+# from airflow.www.fab_security.manager import AUTH_REMOTE_USER  # type: ignore
 
 
 basedir = os.path.abspath(os.path.dirname(__file__))
diff --git a/airflow/dag_processing/processor.py b/airflow/dag_processing/processor.py
index e59c818f20534..73203ce0e44f2 100644
--- a/airflow/dag_processing/processor.py
+++ b/airflow/dag_processing/processor.py
@@ -424,7 +424,7 @@ def manage_slas(self, dag: DAG, session: Session = None) -> None:
             if next_info is None:
                 self.log.info("Skipping SLA check for %s because task does not have scheduled date", ti)
             else:
-                while next_info.logical_date < ts:
+                while next_info and next_info.logical_date < ts:
                     next_info = dag.next_dagrun_info(next_info.data_interval, restricted=False)
 
                     if next_info is None:
diff --git a/airflow/decorators/base.py b/airflow/decorators/base.py
index 229a114fc9cfb..6df3cb79fb105 100644
--- a/airflow/decorators/base.py
+++ b/airflow/decorators/base.py
@@ -116,7 +116,7 @@ def __init__(
         op_args: Tuple[Any],
         op_kwargs: Dict[str, Any],
         multiple_outputs: bool = False,
-        kwargs_to_upstream: dict = None,
+        kwargs_to_upstream: Optional[dict] = None,
         **kwargs,
     ) -> None:
         kwargs['task_id'] = get_unique_task_id(task_id, kwargs.get('dag'), kwargs.get('task_group'))
@@ -180,7 +180,7 @@ def _hook_apply_defaults(self, *args, **kwargs):
 def task_decorator_factory(
     python_callable: Optional[Callable] = None,
     multiple_outputs: Optional[bool] = None,
-    decorated_operator_class: BaseOperator = None,
+    decorated_operator_class: Optional[BaseOperator] = None,
     **kwargs,
 ) -> Callable[[T], T]:
     """
diff --git a/airflow/executors/celery_executor.py b/airflow/executors/celery_executor.py
index 5f47115d3eadc..1b246a2b3f806 100644
--- a/airflow/executors/celery_executor.py
+++ b/airflow/executors/celery_executor.py
@@ -106,6 +106,16 @@ def _execute_in_fork(command_to_exec: CommandType, celery_task_id: Optional[str]
     try:
         from airflow.cli.cli_parser import get_parser
 
+        if not settings.engine:
+            raise AirflowException("The engine should be already set here by start() command. "
+                                   "Likely this is an Airflow error that should be fixed. Please "
+                                   "report it as bug (add all details such as logs etc.) at "
+                                   "https://github.com/apache/airflow/issues/new/choose")
+        if not settings.engine.pool:
+            raise AirflowException("The engine.pool should be already set here by start() command. "
+                                   "Likely this is an Airflow error that should be fixed. Please "
+                                   "report it as bug (add all details such as logs etc.) at "
+                                   "https://github.com/apache/airflow/issues/new/choose")
         settings.engine.pool.dispose()
         settings.engine.dispose()
 
diff --git a/airflow/executors/kubernetes_executor.py b/airflow/executors/kubernetes_executor.py
index cdd62f5d23982..f87add8617022 100644
--- a/airflow/executors/kubernetes_executor.py
+++ b/airflow/executors/kubernetes_executor.py
@@ -433,7 +433,7 @@ def __init__(self):
         self.kube_client: Optional[client.CoreV1Api] = None
         self.scheduler_job_id: Optional[str] = None
         self.event_scheduler: Optional[EventScheduler] = None
-        self.last_handled: Dict[TaskInstanceKey, int] = {}
+        self.last_handled: Dict[TaskInstanceKey, float] = {}
         super().__init__(parallelism=self.kube_config.parallelism)
 
     @provide_session
@@ -620,6 +620,11 @@ def sync(self) -> None:
             except Empty:
                 break
 
+        if not self.event_scheduler:
+            raise AirflowException("The event_scheduler should be already set here by start() command. "
+                                   "Likely this is an Airflow error that should be fixed. Please "
+                                   "report it as bug (add all details such as logs etc.) at "
+                                   "https://github.com/apache/airflow/issues/new/choose")
         # Run any pending timed events
         next_event = self.event_scheduler.run(blocking=False)
         self.log.debug("Next timed event is in %f", next_event)
diff --git a/airflow/hooks/subprocess.py b/airflow/hooks/subprocess.py
index f04b2e70ead1d..5db841eae8c97 100644
--- a/airflow/hooks/subprocess.py
+++ b/airflow/hooks/subprocess.py
@@ -31,7 +31,7 @@ class SubprocessHook(BaseHook):
     """Hook for running processes with the ``subprocess`` module"""
 
     def __init__(self) -> None:
-        self.sub_process = None
+        self.sub_process: Optional[Popen] = None
         super().__init__()
 
     def run_command(
@@ -39,7 +39,7 @@ def run_command(
         command: List[str],
         env: Optional[Dict[str, str]] = None,
         output_encoding: str = 'utf-8',
-        cwd: str = None,
+        cwd: Optional[str] = None,
     ) -> SubprocessResult:
         """
         Execute the command.
@@ -84,15 +84,14 @@ def pre_exec():
 
             self.log.info('Output:')
             line = ''
-            for raw_line in iter(self.sub_process.stdout.readline, b''):
+            for raw_line in iter(self.sub_process.stdout.readline, b''):  # type: ignore
                 line = raw_line.decode(output_encoding).rstrip()
                 self.log.info("%s", line)
 
             self.sub_process.wait()
 
             self.log.info('Command exited with return code %s', self.sub_process.returncode)
-
-        return SubprocessResult(exit_code=self.sub_process.returncode, output=line)
+        return SubprocessResult(exit_code=self.sub_process.returncode, output=line)  # type: ignore
 
     def send_sigterm(self):
         """Sends SIGTERM signal to ``self.sub_process`` if one exists."""
diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
index 43eca4ae82cb5..c5956f232f7fa 100644
--- a/airflow/jobs/local_task_job.py
+++ b/airflow/jobs/local_task_job.py
@@ -152,6 +152,11 @@ def handle_task_exit(self, return_code: int) -> None:
         # Without setting this, heartbeat may get us
         self.terminating = True
         self.log.info("Task exited with return code %s", return_code)
+        if not self.task_instance:
+            raise AirflowException("The task_instance should be already set here by start() command. "
+                                   "Likely this is an Airflow error that should be fixed. Please "
+                                   "report it as bug (add all details such as logs etc.) at "
+                                   "https://github.com/apache/airflow/issues/new/choose")
         self.task_instance.refresh_from_db()
 
         if self.task_instance.state == State.RUNNING:
diff --git a/airflow/kubernetes/kubernetes_helper_functions.py b/airflow/kubernetes/kubernetes_helper_functions.py
index bc68daf404149..1b023174702e2 100644
--- a/airflow/kubernetes/kubernetes_helper_functions.py
+++ b/airflow/kubernetes/kubernetes_helper_functions.py
@@ -62,7 +62,7 @@ def annotations_to_key(annotations: Dict[str, str]) -> Optional[TaskInstanceKey]
     dag_id = annotations['dag_id']
     task_id = annotations['task_id']
     try_number = int(annotations['try_number'])
-    run_id = annotations.get('run_id')
+    run_id: Optional[str] = annotations.get('run_id')
     if not run_id and 'execution_date' in annotations:
         # Compat: Look up the run_id from the TI table!
         from airflow.models.dagrun import DagRun
@@ -71,9 +71,9 @@ def annotations_to_key(annotations: Dict[str, str]) -> Optional[TaskInstanceKey]
 
         execution_date = pendulum.parse(annotations['execution_date'])
         # Do _not_ use create-session, we don't want to expunge
-        session = Session()
+        session = Session()  # type: ignore
 
-        run_id: str = (
+        run_id = (
             session.query(TaskInstance.run_id)
             .join(TaskInstance.dag_run)
             .filter(
@@ -84,4 +84,8 @@ def annotations_to_key(annotations: Dict[str, str]) -> Optional[TaskInstanceKey]
             .scalar()
         )
 
+    if not run_id:
+        # TODO: Is this the right way of handling it?
+        raise ValueError("Run Id has to be defined either by `run_id` annotation "
+                         "or derived from `execution_Date")
     return TaskInstanceKey(dag_id, task_id, run_id, try_number)
diff --git a/airflow/kubernetes/pod_generator.py b/airflow/kubernetes/pod_generator.py
index 1499ebdfcd855..03aaed7d358d3 100644
--- a/airflow/kubernetes/pod_generator.py
+++ b/airflow/kubernetes/pod_generator.py
@@ -337,7 +337,7 @@ def construct_pod(
         pod_override_object: Optional[k8s.V1Pod],
         base_worker_pod: k8s.V1Pod,
         namespace: str,
-        scheduler_job_id: int,
+        scheduler_job_id: str,
         run_id: Optional[str] = None,
     ) -> k8s.V1Pod:
         """
@@ -440,7 +440,7 @@ def deserialize_model_dict(pod_dict: dict) -> k8s.V1Pod:
         return api_client._ApiClient__deserialize_model(pod_dict, k8s.V1Pod)
 
     @staticmethod
-    def make_unique_pod_id(pod_id: str) -> str:
+    def make_unique_pod_id(pod_id: str) -> Optional[str]:
         r"""
         Kubernetes pod names must consist of one or more lowercase
         rfc1035/rfc1123 labels separated by '.' with a maximum length of 253
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 52d73b843ce7f..abdb0a6db29e1 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -44,7 +44,7 @@
     Type,
     Union,
     cast,
-    overload,
+    overload, Sequence,
 )
 
 import jinja2
@@ -61,6 +61,7 @@
 from airflow.compat.functools import cached_property
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, DuplicateTaskIdFound, TaskNotFound
+from airflow.models import SlaMiss
 from airflow.models.base import ID_LEN, Base
 from airflow.models.baseoperator import BaseOperator
 from airflow.models.dagbag import DagBag
@@ -337,7 +338,7 @@ def __init__(
         max_active_tasks: int = conf.getint('core', 'max_active_tasks_per_dag'),
         max_active_runs: int = conf.getint('core', 'max_active_runs_per_dag'),
         dagrun_timeout: Optional[timedelta] = None,
-        sla_miss_callback: Optional[Callable[["DAG", str, str, List[str], List[TaskInstance]], None]] = None,
+        sla_miss_callback: Optional[Callable[["DAG", str, str, Sequence[SlaMiss], Sequence[TaskInstance]], None]] = None,
         default_view: str = conf.get('webserver', 'dag_default_view').lower(),
         orientation: str = conf.get('webserver', 'dag_orientation'),
         catchup: bool = conf.getboolean('scheduler', 'catchup_by_default'),
diff --git a/airflow/operators/bash.py b/airflow/operators/bash.py
index 0c96696c7835e..92ddc81e987fb 100644
--- a/airflow/operators/bash.py
+++ b/airflow/operators/bash.py
@@ -143,7 +143,7 @@ def __init__(
         append_env: bool = False,
         output_encoding: str = 'utf-8',
         skip_exit_code: int = 99,
-        cwd: str = None,
+        cwd: Optional[str] = None,
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
diff --git a/airflow/providers/apache/beam/operators/beam.py b/airflow/providers/apache/beam/operators/beam.py
index cc00e534a2e6d..25d34142cbd64 100644
--- a/airflow/providers/apache/beam/operators/beam.py
+++ b/airflow/providers/apache/beam/operators/beam.py
@@ -19,8 +19,9 @@
 import copy
 from abc import ABCMeta
 from contextlib import ExitStack
-from typing import Callable, List, Optional, Tuple, Union
+from typing import Callable, List, Optional, Tuple, Union, Dict, Any
 
+from airflow import AirflowException
 from airflow.models import BaseOperator
 from airflow.providers.apache.beam.hooks.beam import BeamHook, BeamRunnerType
 from airflow.providers.google.cloud.hooks.dataflow import (
@@ -41,11 +42,16 @@ class BeamDataflowMixin(metaclass=ABCMeta):
     """
 
     dataflow_hook: Optional[DataflowHook]
-    dataflow_config: Optional[DataflowConfiguration]
+    dataflow_config: Union[DataflowConfiguration, Dict[Any, Any]]
+    gcp_conn_id: Optional[str]
+    delegate_to: Optional[str]
 
     def _set_dataflow(
         self, pipeline_options: dict, job_name_variable_key: Optional[str] = None
     ) -> Tuple[str, dict, Callable[[str], None]]:
+        if not isinstance(self.dataflow_config, DataflowConfiguration):
+            raise AirflowException("The self.dataflow_config should be converted to DataflowConfiguration "
+                                   "here")
         self.dataflow_hook = self.__set_dataflow_hook()
         self.dataflow_config.project_id = self.dataflow_config.project_id or self.dataflow_hook.project_id
         dataflow_job_name = self.__get_dataflow_job_name()
@@ -56,8 +62,11 @@ def _set_dataflow(
         return dataflow_job_name, pipeline_options, process_line_callback
 
     def __set_dataflow_hook(self) -> DataflowHook:
+        if not isinstance(self.dataflow_config, DataflowConfiguration):
+            raise AirflowException("The self.dataflow_config should be converted to DataflowConfiguration "
+                                   "here")
         self.dataflow_hook = DataflowHook(
-            gcp_conn_id=self.dataflow_config.gcp_conn_id or self.gcp_conn_id,
+            gcp_conn_id=self.dataflow_config.gcp_conn_id or self.gcp_conn_id,  # type: ignore
             delegate_to=self.dataflow_config.delegate_to or self.delegate_to,
             poll_sleep=self.dataflow_config.poll_sleep,
             impersonation_chain=self.dataflow_config.impersonation_chain,
@@ -68,6 +77,12 @@ def __set_dataflow_hook(self) -> DataflowHook:
         return self.dataflow_hook
 
     def __get_dataflow_job_name(self) -> str:
+        if not isinstance(self.dataflow_config, DataflowConfiguration):
+            raise AirflowException("The self.dataflow_config should be converted to DataflowConfiguration "
+                                   "here")
+        if not self.dataflow_config.job_name:
+            raise AirflowException("The self.dataflow_config.job_name "
+                                   f"should be set here: {self.dataflow_config}")
         return DataflowHook.build_dataflow_job_name(
             self.dataflow_config.job_name, self.dataflow_config.append_job_name
         )
@@ -75,6 +90,9 @@ def __get_dataflow_job_name(self) -> str:
     def __get_dataflow_pipeline_options(
         self, pipeline_options: dict, job_name: str, job_name_key: Optional[str] = None
     ) -> dict:
+        if not isinstance(self.dataflow_config, DataflowConfiguration):
+            raise AirflowException("The self.dataflow_config should be converted to DataflowConfiguration "
+                                   "here")
         pipeline_options = copy.deepcopy(pipeline_options)
         if job_name_key is not None:
             pipeline_options[job_name_key] = job_name
@@ -271,6 +289,9 @@ def execute(self, context):
 
     def on_kill(self) -> None:
         if self.dataflow_hook and self.dataflow_job_id:
+            if not isinstance(self.dataflow_config, DataflowConfiguration):
+                raise AirflowException("The self.dataflow_config should be converted"
+                                       " to DataflowConfiguration here")
             self.log.info('Dataflow job with id: `%s` was requested to be cancelled.', self.dataflow_job_id)
             self.dataflow_hook.cancel_job(
                 job_id=self.dataflow_job_id,
@@ -457,6 +478,9 @@ def execute(self, context):
 
     def on_kill(self) -> None:
         if self.dataflow_hook and self.dataflow_job_id:
+            if not isinstance(self.dataflow_config, DataflowConfiguration):
+                raise AirflowException("The self.dataflow_config should be converted "
+                                       "to DataflowConfiguration here")
             self.log.info('Dataflow job with id: `%s` was requested to be cancelled.', self.dataflow_job_id)
             self.dataflow_hook.cancel_job(
                 job_id=self.dataflow_job_id,
diff --git a/airflow/providers/docker/hooks/docker.py b/airflow/providers/docker/hooks/docker.py
index 49c76c71ad1e8..0f1a0776ebe7e 100644
--- a/airflow/providers/docker/hooks/docker.py
+++ b/airflow/providers/docker/hooks/docker.py
@@ -87,7 +87,7 @@ def get_conn(self) -> APIClient:
         self.__login(client)
         return client
 
-    def __login(self, client) -> int:
+    def __login(self, client) -> None:
         self.log.debug('Logging into Docker')
         try:
             client.login(
diff --git a/airflow/providers/docker/operators/docker.py b/airflow/providers/docker/operators/docker.py
index 17b76115954f6..fa5354c824017 100644
--- a/airflow/providers/docker/operators/docker.py
+++ b/airflow/providers/docker/operators/docker.py
@@ -246,6 +246,8 @@ def get_hook(self) -> DockerHook:
 
         :return: The Docker Hook
         """
+        if not self.docker_conn_id:
+            raise ValueError("docker_conn_id should be set")
         return DockerHook(
             docker_conn_id=self.docker_conn_id,
             base_url=self.docker_url,
@@ -257,7 +259,7 @@ def _run_image(self) -> Optional[str]:
         """Run a Docker container with the provided image"""
         self.log.info('Starting docker container from image %s', self.image)
         if not self.cli:
-            raise Exception("The 'cli' should be initialized before!")
+            raise AirflowException("The 'cli' should be initialized before!")
         if self.mount_tmp_dir:
             with TemporaryDirectory(prefix='airflowtmp', dir=self.host_tmp_dir) as host_tmp_dir_generated:
                 tmp_mount = Mount(self.tmp_dir, host_tmp_dir_generated, "bind")
@@ -281,6 +283,8 @@ def _run_image_with_mounts(self, target_mounts, add_tmp_variable: bool) -> Optio
             self.environment['AIRFLOW_TMP_DIR'] = self.tmp_dir
         else:
             self.environment.pop('AIRFLOW_TMP_DIR', None)
+        if not self.cli:
+            raise AirflowException("The 'cli' should be initialized before!")
         self.container = self.cli.create_container(
             command=self.format_command(self.command),
             name=self.container_name,
@@ -304,12 +308,14 @@ def _run_image_with_mounts(self, target_mounts, add_tmp_variable: bool) -> Optio
             working_dir=self.working_dir,
             tty=self.tty,
         )
+        if not self.container:
+            raise AirflowException("This should not happen. The self.container should be set here")
         lines = self.cli.attach(container=self.container['Id'], stdout=True, stderr=True, stream=True)
         try:
             self.cli.start(self.container['Id'])
 
-            line = ''
-            res_lines = []
+            line: Union[str, bytes] = ''
+            res_lines: List[str] = []
             return_value = None
             for line in lines:
                 if hasattr(line, 'decode'):
@@ -320,8 +326,9 @@ def _run_image_with_mounts(self, target_mounts, add_tmp_variable: bool) -> Optio
                 self.log.info(line)
             result = self.cli.wait(self.container['Id'])
             if result['StatusCode'] != 0:
-                res_lines = "\n".join(res_lines)
-                raise AirflowException('docker container failed: ' + repr(result) + f"lines {res_lines}")
+                joined_lines = '\n'.join(res_lines)
+                raise AirflowException('docker container failed: ' + repr(result) +
+                                       f"lines {joined_lines}")
             if self.retrieve_output and not return_value:
                 return_value = self._attempt_to_retrieve_result()
             ret = None
@@ -397,7 +404,7 @@ def _get_cli(self) -> APIClient:
             return APIClient(base_url=self.docker_url, version=self.api_version, tls=tls_config)
 
     @staticmethod
-    def format_command(command: Union[str, List[str]]) -> Union[List[str], str]:
+    def format_command(command: Union[str, List[str], None]) -> Union[List[str], str, None]:
         """
         Retrieve command(s). if command string starts with [, it returns the command list)
 
diff --git a/airflow/providers/elasticsearch/log/es_task_handler.py b/airflow/providers/elasticsearch/log/es_task_handler.py
index cd0897153dfdc..13dc6892111d8 100644
--- a/airflow/providers/elasticsearch/log/es_task_handler.py
+++ b/airflow/providers/elasticsearch/log/es_task_handler.py
@@ -22,7 +22,7 @@
 from datetime import datetime
 from operator import attrgetter
 from time import time
-from typing import List, Optional, Tuple
+from typing import List, Optional, Tuple, Union
 from urllib.parse import quote
 
 # Using `from elasticsearch import *` would break elasticsearch mocking used in unit test.
@@ -97,7 +97,7 @@ def __init__(
         self.json_fields = [label.strip() for label in json_fields.split(",")]
         self.host_field = host_field
         self.offset_field = offset_field
-        self.handler = None
+        self.handler: Optional[Union[logging.StreamHandler, logging.FileHandler]] = None  # type: ignore
         self.context_set = False
 
     def _render_log_id(self, ti: TaskInstance, try_number: int) -> str:
@@ -275,7 +275,7 @@ def set_context(self, ti: TaskInstance) -> None:
         """
         self.mark_end_on_close = not ti.raw
 
-        if self.json_format:
+        if self.json_format and self.formatter:
             self.formatter = JSONFormatter(
                 fmt=self.formatter._fmt,
                 json_fields=self.json_fields + [self.offset_field],
@@ -294,9 +294,9 @@ def set_context(self, ti: TaskInstance) -> None:
                 # already been initialized
                 return
 
-            self.handler = logging.StreamHandler(stream=sys.__stdout__)  # type: ignore
-            self.handler.setLevel(self.level)  # type: ignore
-            self.handler.setFormatter(self.formatter)  # type: ignore
+            self.handler = logging.StreamHandler(stream=sys.__stdout__)
+            self.handler.setLevel(self.level)
+            self.handler.setFormatter(self.formatter)
         else:
             super().set_context(ti)
         self.context_set = True
@@ -320,8 +320,9 @@ def close(self) -> None:
 
         # Reopen the file stream, because FileHandler.close() would be called
         # first in logging.shutdown() and the stream in it would be set to None.
-        if self.handler.stream is None or self.handler.stream.closed:
-            self.handler.stream = self.handler._open()
+        if isinstance(self.handler, logging.FileHandler):
+            if self.handler.stream is None or self.handler.stream.closed:  # type: ignore
+                self.handler.stream = self.handler._open()
 
         # Mark the end of file using end of log mark,
         # so we know where to stop while auto-tailing.
diff --git a/airflow/providers/google/__init__.py b/airflow/providers/google/__init__.py
index dc2ff2a692e50..ce6655105fa75 100644
--- a/airflow/providers/google/__init__.py
+++ b/airflow/providers/google/__init__.py
@@ -19,7 +19,7 @@
 
 # HACK:
 # Sphinx-autoapi doesn't like imports to excluded packages in the main module.
-conf = importlib.import_module('airflow.configuration').conf
+conf = importlib.import_module('airflow.configuration').conf  # type: ignore
 
 PROVIDERS_GOOGLE_VERBOSE_LOGGING: bool = conf.getboolean(
     'providers_google', 'VERBOSE_LOGGING', fallback=False
diff --git a/airflow/providers/google/cloud/_internal_client/secret_manager_client.py b/airflow/providers/google/cloud/_internal_client/secret_manager_client.py
index 504208029e27c..14117be0195df 100644
--- a/airflow/providers/google/cloud/_internal_client/secret_manager_client.py
+++ b/airflow/providers/google/cloud/_internal_client/secret_manager_client.py
@@ -21,7 +21,7 @@
 import google
 
 try:
-    from functools import cached_property
+    from functools import cached_property  # type: ignore
 except ImportError:
     from cached_property import cached_property
 from google.api_core.exceptions import InvalidArgument, NotFound, PermissionDenied
diff --git a/airflow/providers/samba/hooks/samba.py b/airflow/providers/samba/hooks/samba.py
index df46c0cf3489a..8780f2809d112 100644
--- a/airflow/providers/samba/hooks/samba.py
+++ b/airflow/providers/samba/hooks/samba.py
@@ -19,7 +19,7 @@
 import posixpath
 from functools import wraps
 from shutil import copyfileobj
-from typing import Optional
+from typing import Optional, Dict, Any
 
 import smbclient
 
@@ -57,12 +57,12 @@ def __init__(self, samba_conn_id: str = default_conn_name, share: Optional[str]
 
         self._host = conn.host
         self._share = share or conn.schema
-        self._connection_cache = connection_cache = {}
+        self._connection_cache: Dict[Any, Any] = {}
         self._conn_kwargs = {
             "username": conn.login,
             "password": conn.password,
             "port": conn.port or 445,
-            "connection_cache": connection_cache,
+            "connection_cache": self._connection_cache,
         }
 
     def __enter__(self):
diff --git a/airflow/providers/ssh/hooks/ssh.py b/airflow/providers/ssh/hooks/ssh.py
index cc4ef1ca48f1f..c5d32dcd7bb38 100644
--- a/airflow/providers/ssh/hooks/ssh.py
+++ b/airflow/providers/ssh/hooks/ssh.py
@@ -20,10 +20,12 @@
 import warnings
 from base64 import decodebytes
 from io import StringIO
-from typing import Dict, Optional, Tuple, Union
+from typing import Dict, Optional, Tuple, Union, Type
 
 import paramiko
+from paramiko.client import SSHClient
 from paramiko.config import SSH_PORT
+from paramiko.pkey import PKey
 from sshtunnel import SSHTunnelForwarder
 
 from airflow.exceptions import AirflowException
@@ -71,7 +73,7 @@ class SSHHook(BaseHook):
     """
 
     # List of classes to try loading private keys as, ordered (roughly) by most common to least common
-    _pkey_loaders = (
+    _pkey_loaders: Tuple[Type[PKey], ...] = (
         paramiko.RSAKey,
         paramiko.ECDSAKey,
         paramiko.Ed25519Key,
@@ -133,7 +135,7 @@ def __init__(
         self.look_for_keys = True
 
         # Placeholder for deprecated __enter__
-        self.client = None
+        self.client: Optional[SSHClient] = None
 
         # Use connection to override defaults
         if self.ssh_conn_id is not None:
@@ -241,12 +243,16 @@ def __init__(
             with open(user_ssh_config_filename) as config_fd:
                 ssh_conf.parse(config_fd)
             host_info = ssh_conf.lookup(self.remote_host)
-            if host_info and host_info.get('proxycommand'):
-                self.host_proxy = paramiko.ProxyCommand(host_info.get('proxycommand'))
+            if host_info:
+                proxy_command = host_info.get('proxycommand')
+                if proxy_command:
+                    self.host_proxy = paramiko.ProxyCommand(proxy_command)
 
             if not (self.password or self.key_file):
-                if host_info and host_info.get('identityfile'):
-                    self.key_file = host_info.get('identityfile')[0]
+                if host_info:
+                    identity_file = host_info.get('identityfile')
+                    if identity_file:
+                        self.key_file = identity_file[0]
 
         self.port = self.port or SSH_PORT
 
@@ -273,12 +279,13 @@ def get_conn(self) -> paramiko.SSHClient:
         else:
             if self.host_key is not None:
                 client_host_keys = client.get_host_keys()
-                if self.port == SSH_PORT:
-                    client_host_keys.add(self.remote_host, self.host_key.get_name(), self.host_key)
-                else:
-                    client_host_keys.add(
-                        f"[{self.remote_host}]:{self.port}", self.host_key.get_name(), self.host_key
-                    )
+                if self.remote_host:
+                    if self.port == SSH_PORT:
+                        client_host_keys.add(self.remote_host, self.host_key.get_name(), self.host_key)
+                    else:
+                        client_host_keys.add(
+                            f"[{self.remote_host}]:{self.port}", self.host_key.get_name(), self.host_key
+                        )
             else:
                 pass  # will fallback to system host keys if none explicitly specified in conn extra
 
@@ -297,15 +304,17 @@ def get_conn(self) -> paramiko.SSHClient:
             connect_kwargs.update(password=password)
 
         if self.pkey:
-            connect_kwargs.update(pkey=self.pkey)
+            connect_kwargs.update(pkey=self.pkey)  # type: ignore
 
         if self.key_file:
             connect_kwargs.update(key_filename=self.key_file)
 
-        client.connect(**connect_kwargs)
+        client.connect(**connect_kwargs)  # type: ignore
 
         if self.keepalive_interval:
-            client.get_transport().set_keepalive(self.keepalive_interval)
+            transport = client.get_transport()
+            if transport:
+                transport.set_keepalive(self.keepalive_interval)
 
         self.client = client
         return client
diff --git a/airflow/providers/ssh/operators/ssh.py b/airflow/providers/ssh/operators/ssh.py
index c0953222aec90..197783c57086e 100644
--- a/airflow/providers/ssh/operators/ssh.py
+++ b/airflow/providers/ssh/operators/ssh.py
@@ -206,7 +206,7 @@ def run_ssh_client_command(self, ssh_client: SSHClient, command: str) -> bytes:
         return agg_stdout
 
     def execute(self, context=None) -> Union[bytes, str]:
-        result = None
+        result: Optional[Union[bytes, str]] = None
         if self.command is None:
             raise AirflowException("SSH operator error: SSH command not specified. Aborting.")
 
diff --git a/airflow/providers_manager.py b/airflow/providers_manager.py
index 7b47a9a05591e..3a01c040ce67e 100644
--- a/airflow/providers_manager.py
+++ b/airflow/providers_manager.py
@@ -242,7 +242,7 @@ def __init__(self):
         # Keeps dict of hooks keyed by connection type
         self._hooks_dict: Dict[str, HookInfo] = {}
 
-        self._taskflow_decorators: Dict[str, Callable] = LazyDictWithCache()
+        self._taskflow_decorators: LazyDictWithCache[str, Callable] = LazyDictWithCache()
         # keeps mapping between connection_types and hook class, package they come from
         self._hook_provider_dict: Dict[str, HookClassProvider] = {}
         # Keeps dict of hooks keyed by connection type. They are lazy evaluated at access time
@@ -367,7 +367,7 @@ def _discover_all_airflow_builtin_providers_from_local_sources(self) -> None:
             log.info("You have no providers installed.")
             return
         try:
-            for path in airflow.providers.__path__:
+            for path in airflow.providers.__path__:  # type: ignore
                 self._add_provider_info_from_local_source_files_on_path(path)
         except Exception as e:
             log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
@@ -561,7 +561,7 @@ def _add_taskflow_decorator(self, name, decorator_class_name: str, provider_pack
         if name in self._taskflow_decorators:
             try:
                 existing = self._taskflow_decorators[name]
-                other_name = f'{existing.__module__}.{existing.__name}'
+                other_name = f'{existing.__module__}.{existing.__name}'  # type: ignore
             except Exception:
                 # If problem importing, then get the value from the functools.partial
                 other_name = self._taskflow_decorators._raw_dict[name].args[0]
@@ -584,7 +584,10 @@ def _get_attr(obj: Any, attr_name: str):
         return getattr(obj, attr_name)
 
     def _import_hook(
-        self, connection_type: Optional[str], hook_class_name: str = None, package_name: str = None
+        self,
+        connection_type: Optional[str],
+        hook_class_name: Optional[str] = None,
+        package_name: Optional[str] = None,
     ) -> Optional[HookInfo]:
         """
         Imports hook and retrieves hook information. Either connection_type (for lazy loading)
@@ -613,8 +616,10 @@ def _import_hook(
         else:
             if not package_name:
                 raise ValueError(
-                    f"Provider package name is not set when hook_class_name ({hook_class_name}) " f"is used"
+                    f"Provider package name is not set when hook_class_name ({hook_class_name}) is used"
                 )
+        if not hook_class_name:
+            raise ValueError(f"The Hook class is missing for {connection_type} connection type")
         allowed_field_classes = [IntegerField, PasswordField, StringField, BooleanField]
         if not _sanity_check(package_name, hook_class_name):
             return None
@@ -787,7 +792,7 @@ def hooks(self) -> Dict[str, Optional[HookInfo]]:
         return self._hooks_lazy_dict  # type: ignore
 
     @property
-    def taskflow_decorators(self) -> Dict[str, Callable]:
+    def taskflow_decorators(self) -> LazyDictWithCache:
         self.initialize_providers_taskflow_decorator()
         return self._taskflow_decorators
 
diff --git a/airflow/settings.py b/airflow/settings.py
index d7499ad0998bf..9f470a580d7eb 100644
--- a/airflow/settings.py
+++ b/airflow/settings.py
@@ -22,7 +22,7 @@
 import os
 import sys
 import warnings
-from typing import Optional
+from typing import Optional, List
 
 import pendulum
 import sqlalchemy
@@ -564,7 +564,7 @@ def initialize():
 #   ]
 #
 # DASHBOARD_UIALERTS: List["UIAlert"]
-DASHBOARD_UIALERTS = []
+DASHBOARD_UIALERTS: List[str] = []
 
 # Prefix used to identify tables holding data moved during migration.
 AIRFLOW_MOVED_TABLE_PREFIX = "_airflow_moved"
diff --git a/airflow/timetables/base.py b/airflow/timetables/base.py
index e97f2532b9e1b..850bc4711389d 100644
--- a/airflow/timetables/base.py
+++ b/airflow/timetables/base.py
@@ -33,7 +33,7 @@ class DataInterval(NamedTuple):
     end: DateTime
 
     @classmethod
-    def exact(cls, at: DateTime) -> "DagRunInfo":
+    def exact(cls, at: DateTime) -> "DataInterval":
         """Represent an "interval" containing only an exact time."""
         return cls(start=at, end=at)
 
diff --git a/airflow/timetables/interval.py b/airflow/timetables/interval.py
index d669cb652d153..9018f6caff705 100644
--- a/airflow/timetables/interval.py
+++ b/airflow/timetables/interval.py
@@ -265,7 +265,7 @@ def serialize(self) -> Dict[str, Any]:
         from airflow.serialization.serialized_objects import encode_relativedelta
 
         if isinstance(self._delta, datetime.timedelta):
-            delta = self._delta.total_seconds()
+            delta: Any = self._delta.total_seconds()
         else:
             delta = encode_relativedelta(self._delta)
         return {"delta": delta}
diff --git a/airflow/triggers/temporal.py b/airflow/triggers/temporal.py
index 7cd67e21f200c..9b4d21dca4b31 100644
--- a/airflow/triggers/temporal.py
+++ b/airflow/triggers/temporal.py
@@ -38,7 +38,7 @@ def __init__(self, moment: datetime.datetime):
         # Make sure it's in UTC
         elif moment.tzinfo is None:
             raise ValueError("You cannot pass naive datetimes")
-        elif not hasattr(moment.tzinfo, "offset") or moment.tzinfo.offset != 0:
+        elif not hasattr(moment.tzinfo, "offset") or moment.tzinfo.offset != 0:  # type: ignore
             raise ValueError(f"The passed datetime must be using Pendulum's UTC, not {moment.tzinfo!r}")
         else:
             self.moment = moment
diff --git a/airflow/typing_compat.py b/airflow/typing_compat.py
index 26237c20eaeda..9f1185f76c295 100644
--- a/airflow/typing_compat.py
+++ b/airflow/typing_compat.py
@@ -36,4 +36,4 @@
 except ImportError:
     import re
 
-    RePatternType = type(re.compile('', 0))
+    RePatternType = type(re.compile('', 0))  # type: ignore
diff --git a/airflow/utils/db.py b/airflow/utils/db.py
index be74b27646f15..4abeb5c2b67d0 100644
--- a/airflow/utils/db.py
+++ b/airflow/utils/db.py
@@ -836,7 +836,7 @@ def check_task_tables_without_matching_dagruns(session) -> Iterable[str]:
     for model in models_to_dagrun:
         # We can't use the model here since it may differ from the db state due to
         # this function is run prior to migration. Use the reflected table instead.
-        source_table = metadata.tables.get(model.__tablename__)
+        source_table = metadata.tables.get(model.__tablename__)  # type: ignore
         if source_table is None:
             continue
 
@@ -889,7 +889,7 @@ def _check_migration_errors(session=None) -> Iterable[str]:
     ):
         yield from check_fn(session)
         # Ensure there is no "active" transaction. Seems odd, but without this MSSQL can hang
-        session.commit()
+        session.commit()  # type: ignore
 
 
 @provide_session
diff --git a/airflow/utils/dot_renderer.py b/airflow/utils/dot_renderer.py
index 799e1f808d4cd..1ab7be23e8f73 100644
--- a/airflow/utils/dot_renderer.py
+++ b/airflow/utils/dot_renderer.py
@@ -17,10 +17,11 @@
 # specific language governing permissions and limitations
 # under the License.
 """Renderer DAG (tasks and dependencies) to the graphviz object."""
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Any
 
 import graphviz
 
+from airflow import AirflowException
 from airflow.models import TaskInstance
 from airflow.models.baseoperator import BaseOperator
 from airflow.models.dag import DAG
@@ -46,7 +47,7 @@ def _refine_color(color: str):
     return color
 
 
-def _draw_task(task: BaseOperator, parent_graph: graphviz.Digraph, states_by_task_id: Dict[str, str]) -> None:
+def _draw_task(task: BaseOperator, parent_graph: graphviz.Digraph, states_by_task_id: Optional[Dict[Any, Any]]) -> None:
     """Draw a single task on the given parent_graph"""
     if states_by_task_id:
         state = states_by_task_id.get(task.task_id, State.NONE)
@@ -69,7 +70,7 @@ def _draw_task(task: BaseOperator, parent_graph: graphviz.Digraph, states_by_tas
 
 
 def _draw_task_group(
-    task_group: TaskGroup, parent_graph: graphviz.Digraph, states_by_task_id: Dict[str, str]
+    task_group: TaskGroup, parent_graph: graphviz.Digraph, states_by_task_id: Optional[Dict[str, str]]
 ) -> None:
     """Draw the given task_group and its children on the given parent_graph"""
     # Draw joins
@@ -102,15 +103,18 @@ def _draw_task_group(
         )
 
     # Draw children
-    for child in sorted(task_group.children.values(), key=lambda t: t.label):
+    for child in sorted(task_group.children.values(), key=lambda t: t.label):  # type: ignore
         _draw_nodes(child, parent_graph, states_by_task_id)
 
 
-def _draw_nodes(node: TaskMixin, parent_graph: graphviz.Digraph, states_by_task_id: Dict[str, str]) -> None:
+def _draw_nodes(node: TaskMixin, parent_graph: graphviz.Digraph,
+                states_by_task_id: Optional[Dict[Any, Any]]) -> None:
     """Draw the node and its children on the given parent_graph recursively."""
     if isinstance(node, BaseOperator):
         _draw_task(node, parent_graph, states_by_task_id)
     else:
+        if not isinstance(node, TaskGroup):
+            raise AirflowException(f"The node {node} should be TaskGroup and is not")
         # Draw TaskGroup
         if node.is_root:
             # No need to draw background for root TaskGroup.
diff --git a/airflow/utils/edgemodifier.py b/airflow/utils/edgemodifier.py
index 9fbac62b9d2ad..f3ce6a8431811 100644
--- a/airflow/utils/edgemodifier.py
+++ b/airflow/utils/edgemodifier.py
@@ -15,8 +15,9 @@
 # specific language governing permissions and limitations
 # under the License.
 
-from typing import Sequence, Union
+from typing import List, Optional, Sequence, Union
 
+from airflow.models import BaseOperator
 from airflow.models.taskmixin import TaskMixin
 
 
@@ -38,10 +39,10 @@ class EdgeModifier(TaskMixin):
     is the representation of the information for one specific edge.
     """
 
-    def __init__(self, label: str = None):
+    def __init__(self, label: Optional[str] = None):
         self.label = label
-        self._upstream = []
-        self._downstream = []
+        self._upstream: List[BaseOperator] = []
+        self._downstream: List[BaseOperator] = []
 
     @property
     def roots(self):
@@ -61,7 +62,7 @@ def set_upstream(self, task_or_task_list: Union[TaskMixin, Sequence[TaskMixin]],
         Providing this also provides << via TaskMixin.
         """
         # Ensure we have a list, even if it's just one item
-        if not isinstance(task_or_task_list, list):
+        if isinstance(task_or_task_list, TaskMixin):
             task_or_task_list = [task_or_task_list]
         # Unfurl it into actual operators
         operators = []
@@ -85,7 +86,7 @@ def set_downstream(self, task_or_task_list: Union[TaskMixin, Sequence[TaskMixin]
         Providing this also provides >> via TaskMixin.
         """
         # Ensure we have a list, even if it's just one item
-        if not isinstance(task_or_task_list, list):
+        if isinstance(task_or_task_list, TaskMixin):
             task_or_task_list = [task_or_task_list]
         # Unfurl it into actual operators
         operators = []
diff --git a/airflow/utils/entry_points.py b/airflow/utils/entry_points.py
index 8e14b5eaddce8..668ed9b9941fb 100644
--- a/airflow/utils/entry_points.py
+++ b/airflow/utils/entry_points.py
@@ -18,7 +18,7 @@
 try:
     import importlib_metadata
 except ImportError:
-    from importlib import metadata as importlib_metadata
+    from importlib import metadata as importlib_metadata  # type: ignore
 
 
 def entry_points_with_dist(group: str):
diff --git a/airflow/utils/json.py b/airflow/utils/json.py
index d859fd1b8fa47..9fc649598201d 100644
--- a/airflow/utils/json.py
+++ b/airflow/utils/json.py
@@ -24,7 +24,7 @@
 try:
     import numpy as np
 except ImportError:
-    np = None
+    np = None  # type: ignore
 
 try:
     from kubernetes.client import models as k8s
diff --git a/airflow/utils/log/logging_mixin.py b/airflow/utils/log/logging_mixin.py
index 1fc075724223f..bdf0ef7d8b947 100644
--- a/airflow/utils/log/logging_mixin.py
+++ b/airflow/utils/log/logging_mixin.py
@@ -19,9 +19,13 @@
 import logging
 import re
 import sys
+from io import IOBase
 from logging import Handler, Logger, StreamHandler
 
 # 7-bit C1 ANSI escape sequences
+from types import TracebackType
+from typing import ContextManager, IO, Optional, Type, AnyStr, Iterator, Iterable
+
 ANSI_ESCAPE = re.compile(r'\x1B[@-_][0-?]*[ -/]*[@-~]')
 
 
@@ -72,8 +76,7 @@ def supports_external_link(self) -> bool:
         """Return whether handler is able to support external links."""
 
 
-# TODO: Formally inherit from io.IOBase
-class StreamLogWriter:
+class StreamLogWriter(IO[str]):
     """Allows to redirect stdout and stderr to logger"""
 
     encoding: None = None
@@ -134,6 +137,52 @@ def isatty(self):
         """
         return False
 
+    def fileno(self) -> int:
+        pass
+
+    def read(self, n: int = ...) -> AnyStr:
+        pass
+
+    def readable(self) -> bool:
+        pass
+
+    def readline(self, limit: int = ...) -> AnyStr:
+        pass
+
+    def readlines(self, hint: int = ...) -> list:
+        pass
+
+    def seek(self, offset: int, whence: int = ...) -> int:
+        pass
+
+    def seekable(self) -> bool:
+        pass
+
+    def tell(self) -> int:
+        pass
+
+    def truncate(self, size: Optional[int] = ...) -> int:
+        pass
+
+    def writable(self) -> bool:
+        pass
+
+    def writelines(self, lines: Iterable[AnyStr]) -> None:
+        pass
+
+    def __next__(self) -> AnyStr:
+        pass
+
+    def __iter__(self) -> Iterator[AnyStr]:
+        pass
+
+    def __enter__(self) -> IO[AnyStr]:
+        pass
+
+    def __exit__(self, t: Optional[Type[BaseException]], value: Optional[BaseException],
+                 traceback: Optional[TracebackType]) -> Optional[bool]:
+        pass
+
 
 class RedirectStdHandler(StreamHandler):
     """
diff --git a/airflow/utils/log/secrets_masker.py b/airflow/utils/log/secrets_masker.py
index 6607c399cf9d7..db3874e34f840 100644
--- a/airflow/utils/log/secrets_masker.py
+++ b/airflow/utils/log/secrets_masker.py
@@ -18,14 +18,14 @@
 import collections
 import logging
 import re
-from typing import TYPE_CHECKING, Iterable, Optional, Set, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Set, Tuple, TypeVar, Union
 
 from airflow.compat.functools import cache, cached_property
 
 if TYPE_CHECKING:
     from airflow.typing_compat import RePatternType
 
-    RedactableItem = TypeVar('RedactableItem')
+    RedactableItem = TypeVar('RedactableItem', str, Dict[Any, Any], Tuple[Any, ...], List[Any])
 
 
 log = logging.getLogger(__name__)
@@ -72,7 +72,7 @@ def should_hide_value_for_key(name):
     return False
 
 
-def mask_secret(secret: Union[str, dict, Iterable], name: str = None) -> None:
+def mask_secret(secret: Union[str, dict, Iterable], name: Optional[str] = None) -> None:
     """
     Mask a secret from appearing in the task logs.
 
@@ -93,7 +93,7 @@ def mask_secret(secret: Union[str, dict, Iterable], name: str = None) -> None:
     _secrets_masker().add_mask(secret, name)
 
 
-def redact(value: "RedactableItem", name: str = None) -> "RedactableItem":
+def redact(value: "RedactableItem", name: Optional[str] = None) -> "RedactableItem":
     """Redact any secrets found in ``value``."""
     return _secrets_masker().redact(value, name)
 
@@ -232,7 +232,7 @@ def redact(self, item: "RedactableItem", name: Optional[str] = None) -> "Redacta
         """
         return self._redact(item, name, depth=0)
 
-    def add_mask(self, secret: Union[str, dict, Iterable], name: str = None):
+    def add_mask(self, secret: Union[str, dict, Iterable], name: Optional[str] = None):
         """Add a new secret to be masked to this filter instance."""
         from airflow.configuration import conf
 
diff --git a/airflow/utils/weekday.py b/airflow/utils/weekday.py
index 3718589aff488..2bfdea5289402 100644
--- a/airflow/utils/weekday.py
+++ b/airflow/utils/weekday.py
@@ -32,7 +32,7 @@ class WeekDay(enum.IntEnum):
     SUNDAY = 7
 
     @classmethod
-    def get_weekday_number(cls, week_day_str):
+    def get_weekday_number(cls, week_day_str: str):
         """
         Return the ISO Week Day Number for a Week Day
 
@@ -55,7 +55,9 @@ def convert(cls, day: Union[str, 'WeekDay']) -> int:
         return cls.get_weekday_number(week_day_str=day)
 
     @classmethod
-    def validate_week_day(cls, week_day: Union[str, 'WeekDay', Set[str], List[str]]):
+    def validate_week_day(
+        cls, week_day: Union[str, 'WeekDay', Set[str], Set['WeekDay'], List[str], List['WeekDay']]
+    ):
         """Validate each item of iterable and create a set to ease compare of values"""
         if not isinstance(week_day, Iterable):
             if isinstance(week_day, WeekDay):
diff --git a/airflow/version.py b/airflow/version.py
index fdd3ccec8ddbe..cd3d0812e845f 100644
--- a/airflow/version.py
+++ b/airflow/version.py
@@ -22,7 +22,7 @@
 try:
     import importlib_metadata as metadata
 except ImportError:
-    from importlib import metadata
+    from importlib import metadata  # type: ignore
 
 try:
     version = metadata.version('apache-airflow')
diff --git a/airflow/www/fab_security/manager.py b/airflow/www/fab_security/manager.py
index 460099c82531e..e5fccbb036141 100644
--- a/airflow/www/fab_security/manager.py
+++ b/airflow/www/fab_security/manager.py
@@ -19,6 +19,10 @@
 # (https://github.com/dpgaspar/Flask-AppBuilder).
 # Copyright 2013, Daniel Vaz Gaspar
 
+# Ignore type checking here, the FAB security manager has been ported from FAB
+# and it has a bit too many class variables/dynamic typing to bother
+# type: ignore
+
 import base64
 import datetime
 import json
diff --git a/airflow/www/fab_security/sqla/manager.py b/airflow/www/fab_security/sqla/manager.py
index 7613a2d64f482..7fa9e3305feec 100644
--- a/airflow/www/fab_security/sqla/manager.py
+++ b/airflow/www/fab_security/sqla/manager.py
@@ -15,6 +15,11 @@
 # specific language governing permissions and limitations
 # under the License.
 
+# Ignore type checking here, the FAB security manager has been ported from FAB
+# and it has a bit too many class variables/dynamic typing to bother
+# type: ignore
+
+
 import logging
 import uuid
 from typing import List, Optional
diff --git a/airflow/www/security.py b/airflow/www/security.py
index 51c357de77d62..8127862de8554 100644
--- a/airflow/www/security.py
+++ b/airflow/www/security.py
@@ -17,6 +17,10 @@
 # under the License.
 #
 
+# Ignore type checking here, the FAB security manager has been ported from FAB
+# and it has a bit too many class variables/dynamic typing to bother
+# type: ignore
+
 import warnings
 from typing import Dict, Optional, Sequence, Set, Tuple
 
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 77c8f62e7bc2a..77bb2acfd2c70 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -26,11 +26,11 @@
 import sys
 import traceback
 from collections import defaultdict
-from datetime import timedelta
+from datetime import timedelta, datetime
 from functools import wraps
 from json import JSONDecodeError
 from operator import itemgetter
-from typing import Any, Callable, List, Optional, Set, Tuple, Union
+from typing import Any, Callable, List, Optional, Set, Tuple, Union, Dict
 from urllib.parse import parse_qsl, unquote, urlencode, urlparse
 
 import lazy_object_proxy
@@ -543,7 +543,7 @@ class AirflowBaseView(BaseView):
     }
 
     if not conf.getboolean('core', 'unit_test_mode'):
-        extra_args['sqlite_warning'] = settings.Session.bind.dialect.name == 'sqlite'
+        extra_args['sqlite_warning'] = settings.Session.bind.dialect.name == 'sqlite'  # type: ignore
         extra_args['sequential_executor_warning'] = conf.get('core', 'executor') == 'SequentialExecutor'
 
     line_chart_attr = {
@@ -4559,8 +4559,8 @@ class DagDependenciesView(AirflowBaseView):
         )
     )
     last_refresh = timezone.utcnow() - refresh_interval
-    nodes = []
-    edges = []
+    nodes: List[Dict[str, Any]] = []
+    edges: List[Dict[str, Any]] = []
 
     @expose('/dag-dependencies')
     @auth.has_access(
@@ -4596,8 +4596,8 @@ def list(self):
 
     def _calculate_graph(self):
 
-        nodes = []
-        edges = []
+        nodes: List[Dict[str, Any]] = []
+        edges: List[Dict[str, Any]] = []
 
         for dag, dependencies in SerializedDagModel.get_dag_dependencies().items():
             dag_node_id = f"dag:{dag}"
diff --git a/dev/import_all_classes.py b/dev/import_all_classes.py
index 54cd7472cb5f6..04ec150aa3478 100755
--- a/dev/import_all_classes.py
+++ b/dev/import_all_classes.py
@@ -22,7 +22,7 @@
 import traceback
 import warnings
 from inspect import isclass
-from typing import List, Set, Tuple
+from typing import List, Set, Tuple, Optional
 from warnings import WarningMessage
 
 from rich import print
@@ -31,7 +31,7 @@
 def import_all_classes(
     paths: List[str],
     prefix: str,
-    provider_ids: List[str] = None,
+    provider_ids: Optional[List[str]] = None,
     print_imports: bool = False,
     print_skips: bool = False,
 ) -> Tuple[List[str], List[WarningMessage]]:
diff --git a/dev/validate_version_added_fields_in_config.py b/dev/validate_version_added_fields_in_config.py
index 20f9b1d03425a..fdd869c7b7e06 100755
--- a/dev/validate_version_added_fields_in_config.py
+++ b/dev/validate_version_added_fields_in_config.py
@@ -96,7 +96,7 @@ def read_local_config_options() -> Set[Tuple[str, str, str]]:
 local_options_plain: Set[Tuple[str, str]] = {
     (section_name, option_name) for section_name, option_name, version_added in local_options
 }
-computed_options: Set[Tuple[str, str, str]] = {
+computed_options = {
     (section_name, option_name, version_added)
     for section_name, option_name, version_added in computed_options
     if (section_name, option_name) in local_options_plain
diff --git a/scripts/ci/pre_commit/pre_commit_check_pre_commit_hook_names.py b/scripts/ci/pre_commit/pre_commit_check_pre_commit_hook_names.py
index 5e16f425a68c6..cbc2d17b920a0 100755
--- a/scripts/ci/pre_commit/pre_commit_check_pre_commit_hook_names.py
+++ b/scripts/ci/pre_commit/pre_commit_check_pre_commit_hook_names.py
@@ -27,7 +27,7 @@
 try:
     from yaml import CSafeLoader as SafeLoader
 except ImportError:
-    from yaml import SafeLoader  # type: ignore[no-redef]
+    from yaml import SafeLoader  # type: ignore
 
 
 def main() -> int:
diff --git a/scripts/ci/pre_commit/pre_commit_check_provider_yaml_files.py b/scripts/ci/pre_commit/pre_commit_check_provider_yaml_files.py
index 6cbd8ab2704d7..ec614a5e466b3 100755
--- a/scripts/ci/pre_commit/pre_commit_check_provider_yaml_files.py
+++ b/scripts/ci/pre_commit/pre_commit_check_provider_yaml_files.py
@@ -23,7 +23,7 @@
 from collections import Counter
 from glob import glob
 from itertools import chain, product
-from typing import Any, Dict, Iterable, List
+from typing import Any, Dict, Iterable, List, Set
 
 import jsonschema
 import yaml
@@ -34,7 +34,7 @@
 try:
     from yaml import CSafeLoader as SafeLoader
 except ImportError:
-    from yaml import SafeLoader  # type: ignore[no-redef]
+    from yaml import SafeLoader  # type: ignore
 
 if __name__ != "__main__":
     raise Exception(
@@ -80,7 +80,7 @@ def _load_package_data(package_paths: Iterable[str]):
     return result
 
 
-def get_all_integration_names(yaml_files):
+def get_all_integration_names(yaml_files) -> List[str]:
     all_integrations = [
         i['integration-name'] for f in yaml_files.values() if 'integrations' in f for i in f["integrations"]
     ]
@@ -137,7 +137,7 @@ def assert_sets_equal(set1, set2):
 
 
 def check_if_objects_belongs_to_package(
-    object_names: List[str], provider_package: str, yaml_file_path: str, resource_type: str
+    object_names: Set[str], provider_package: str, yaml_file_path: str, resource_type: str
 ):
     for object_name in object_names:
         if not object_name.startswith(provider_package):
@@ -283,8 +283,8 @@ def check_invalid_integration(yaml_files: Dict[str, Dict]):
 
 def check_doc_files(yaml_files: Dict[str, Dict]):
     print("Checking doc files")
-    current_doc_urls = []
-    current_logo_urls = []
+    current_doc_urls: List[str] = []
+    current_logo_urls: List[str] = []
     for provider in yaml_files.values():
         if 'integrations' in provider:
             current_doc_urls.extend(
diff --git a/scripts/ci/pre_commit/pre_commit_json_schema.py b/scripts/ci/pre_commit/pre_commit_json_schema.py
index c949ea8bd1e3f..eb52aa0e49a63 100755
--- a/scripts/ci/pre_commit/pre_commit_json_schema.py
+++ b/scripts/ci/pre_commit/pre_commit_json_schema.py
@@ -149,6 +149,8 @@ def _default_validator(validator, default, instance, schema):
 def _load_spec(spec_file: Optional[str], spec_url: Optional[str]):
     if spec_url:
         spec_file = fetch_and_cache(url=spec_url, output_filename=re.sub(r"[^a-zA-Z0-9]", "-", spec_url))
+    if not spec_file:
+        raise Exception(f"The {spec_file} was None and {spec_url} did not lead to any file loading.")
     with open(spec_file) as schema_file:
         schema = json.loads(schema_file.read())
     return schema
diff --git a/scripts/ci/pre_commit/pre_commit_mypy.sh b/scripts/ci/pre_commit/pre_commit_mypy.sh
index 7e2b4f6223f77..5f3198e15b0b4 100755
--- a/scripts/ci/pre_commit/pre_commit_mypy.sh
+++ b/scripts/ci/pre_commit/pre_commit_mypy.sh
@@ -20,8 +20,5 @@ export FORCE_ANSWER_TO_QUESTIONS=${FORCE_ANSWER_TO_QUESTIONS:="quit"}
 export REMEMBER_LAST_ANSWER="true"
 export PRINT_INFO_FROM_SCRIPTS="false"
 
-# Temporarily remove mypy checks until we fix them for Python 3.7
-exit 0
-
 # shellcheck source=scripts/ci/static_checks/mypy.sh
 . "$( dirname "${BASH_SOURCE[0]}" )/../static_checks/mypy.sh" "${@}"
diff --git a/scripts/ci/static_checks/mypy.sh b/scripts/ci/static_checks/mypy.sh
index 7b5879e1e3d13..8cfef967e450f 100755
--- a/scripts/ci/static_checks/mypy.sh
+++ b/scripts/ci/static_checks/mypy.sh
@@ -26,7 +26,7 @@ function run_mypy() {
       files=("$@")
     fi
 
-    docker_v run "${EXTRA_DOCKER_FLAGS[@]}" \
+    docker_v run "${EXTRA_DOCKER_FLAGS[@]}" -t \
         --entrypoint "/usr/local/bin/dumb-init"  \
         "-v" "${AIRFLOW_SOURCES}/.mypy_cache:/opt/airflow/.mypy_cache" \
         "${AIRFLOW_CI_IMAGE_WITH_TAG}" \
diff --git a/scripts/ci/static_checks/run_static_checks.sh b/scripts/ci/static_checks/run_static_checks.sh
index 7f520e0a2d448..fd23938382e02 100755
--- a/scripts/ci/static_checks/run_static_checks.sh
+++ b/scripts/ci/static_checks/run_static_checks.sh
@@ -34,6 +34,13 @@ python -m pip install --user pre-commit \
 
 export PATH=~/.local/bin:${PATH}
 
+if [[ ${DO_NOT_FAIL_ON_ERROR=} != "" ]]; then
+    echo
+    echo "Skip failing on error"
+    echo
+    set +e
+fi
+
 if [[ $# == "0" ]]; then
     pre-commit run --all-files --show-diff-on-failure --color always
 else
@@ -42,3 +49,7 @@ else
         pre-commit run "${pre_commit_check}" --all-files --show-diff-on-failure --color always
     done
 fi
+
+if [[ ${DO_NOT_FAIL_ON_ERROR=} != "" ]]; then
+    exit 0
+fi
diff --git a/scripts/in_container/check_junitxml_result.py b/scripts/in_container/check_junitxml_result.py
index cc69d942f254c..e6971bf7335ff 100755
--- a/scripts/in_container/check_junitxml_result.py
+++ b/scripts/in_container/check_junitxml_result.py
@@ -29,8 +29,11 @@
         with open(fname) as fh:
             root = ET.parse(fh)
         testsuite = root.find('.//testsuite')
-        num_failures = int(testsuite.get('failures'))
-        num_errors = int(testsuite.get('errors'))
+        num_failures = 0
+        num_errors = 0
+        if not testsuite:
+            num_failures = int(testsuite.get('failures'))  # type: ignore
+            num_errors = int(testsuite.get('errors'))  # type: ignore
         if num_failures == 0 and num_errors == 0:
             print(f'\n{TEXT_GREEN}==== No errors, no failures. Good to go! ===={TEXT_RESET}\n')
             sys.exit(0)
diff --git a/scripts/in_container/run_mypy.sh b/scripts/in_container/run_mypy.sh
index fefcb18210b57..3dac50faf7810 100755
--- a/scripts/in_container/run_mypy.sh
+++ b/scripts/in_container/run_mypy.sh
@@ -20,18 +20,4 @@
 . "$( dirname "${BASH_SOURCE[0]}" )/_in_container_script_init.sh"
 export PYTHONPATH=${AIRFLOW_SOURCES}
 
-mypy_args=()
-
-# Mypy doesn't cope very well with namespace packages when give filenames (it gets confused about the lack of __init__.py in airflow.providers, and thinks airflow.providers.docker is the same as a "docker" top level module).
-#
-# So we instead need to convert the file names in to module names to check
-for filename in "$@";
-do
-    if [[ "${filename}" == docs/* ]]; then
-        mypy_args+=("$filename")
-    else
-        mypy_args+=("-m" "$(filename_to_python_module "$filename")")
-    fi
-done
-
-mypy --namespace-packages "${mypy_args[@]}"
+mypy "${@}"
diff --git a/setup.py b/setup.py
index 0490e5d6e3aa1..e6d762b92e232 100644
--- a/setup.py
+++ b/setup.py
@@ -513,6 +513,31 @@ def write_version(filename: str = os.path.join(*[my_dir, "airflow", "git_version
 ]
 # End dependencies group
 
+# Mypy 0.900 and above ships only with stubs from stdlib so if we need other stubs, we need to install them
+# manually as `types-*`. See https://mypy.readthedocs.io/en/stable/running_mypy.html#missing-imports
+# for details. Wy want to install them explicitly because we want to eventually move to
+# mypyd which does not support installing the types dynamically with --install-types
+mypy_dependencies = [
+    'mypy==0.910',
+    'types-croniter',
+    'types-docutils',
+    'types-freezegun',
+    'types-paramiko',
+    'types-protobuf',
+    'types-python-dateutil',
+    'types-python-slugify',
+    'types-pytz',
+    'types-redis',
+    'types-requests',
+    'types-setuptools',
+    'types-termcolor',
+    'types-tabulate',
+    'types-Markdown',
+    'types-PyMySQL',
+    'types-PyYAML',
+]
+
+# Dependencies needed for development only
 devel_only = [
     'aws_xray_sdk',
     'beautifulsoup4~=4.7.1',
@@ -533,7 +558,6 @@ def write_version(filename: str = os.path.join(*[my_dir, "airflow", "git_version
     'jsondiff',
     'mongomock',
     'moto~=2.2, >=2.2.12',
-    'mypy==0.770',
     'parameterized',
     'paramiko',
     'pipdeptree',
@@ -558,7 +582,7 @@ def write_version(filename: str = os.path.join(*[my_dir, "airflow", "git_version
     'yamllint',
 ]
 
-devel = cgroups + devel_only + doc + kubernetes + mysql + pandas + password
+devel = cgroups + devel_only + doc + kubernetes + mypy_dependencies + mysql + pandas + password
 devel_hadoop = devel + hdfs + hive + kerberos + presto + webhdfs
 
 # Dict of all providers which are part of the Apache Airflow repository together with their requirements
@@ -893,6 +917,10 @@ def get_all_provider_packages() -> str:
 class AirflowDistribution(Distribution):
     """The setuptools.Distribution subclass with Airflow specific behaviour"""
 
+    def __init__(self, attrs=None):
+        super().__init__(attrs)
+        self.install_requires = None
+
     def parse_config_files(self, *args, **kwargs) -> None:
         """
         Ensure that when we have been asked to install providers from sources
@@ -989,7 +1017,7 @@ def add_all_provider_packages() -> None:
 class Develop(develop_orig):
     """Forces removal of providers in editable mode."""
 
-    def run(self) -> None:
+    def run(self) -> None:  # type: ignore
         self.announce('Installing in editable mode. Uninstalling provider packages!', level=log.INFO)
         # We need to run "python3 -m pip" because it might be that older PIP binary is in the path
         # And it results with an error when running pip directly (cannot import pip module)
@@ -1052,11 +1080,11 @@ def include_provider_namespace_packages_when_installing_from_sources() -> None:
             'extra_clean': CleanCommand,
             'compile_assets': CompileAssets,
             'list_extras': ListExtras,
-            'install': Install,
+            'install': Install,  # type: ignore
             'develop': Develop,
         },
         test_suite='setup.airflow_test_suite',
-        **setup_kwargs,
+        **setup_kwargs,  # type: ignore
     )
 
 
diff --git a/tests/build_provider_packages_dependencies.py b/tests/build_provider_packages_dependencies.py
index 3ea7bf72f2f38..436425d52b8a2 100644
--- a/tests/build_provider_packages_dependencies.py
+++ b/tests/build_provider_packages_dependencies.py
@@ -75,6 +75,8 @@ def get_provider_from_file_name(file_name: str) -> Optional[str]:
         errors.append(f"Wrong file not in the providers package = {file_name}")
         return None
     suffix = get_file_suffix(file_name)
+    if not suffix:
+        raise Exception(f"The file {file_name} has no suffix!")
     split_path = suffix.split(sep)[2:]
     provider = find_provider(split_path)
     if not provider and file_name.endswith("__init__.py"):
diff --git a/tests/providers/amazon/aws/utils/eks_test_constants.py b/tests/providers/amazon/aws/utils/eks_test_constants.py
index 4624cd40f809d..004fdd74ece8e 100644
--- a/tests/providers/amazon/aws/utils/eks_test_constants.py
+++ b/tests/providers/amazon/aws/utils/eks_test_constants.py
@@ -76,21 +76,21 @@
 class ResponseAttributes:
     """Key names for the dictionaries returned by API calls."""
 
-    CLUSTER: slice = "cluster"
-    CLUSTERS: slice = "clusters"
-    FARGATE_PROFILE_NAMES: slice = "fargateProfileNames"
-    FARGATE_PROFILE: slice = "fargateProfile"
-    NEXT_TOKEN: slice = "nextToken"
-    NODEGROUP: slice = "nodegroup"
-    NODEGROUPS: slice = "nodegroups"
+    CLUSTER: str = "cluster"
+    CLUSTERS: str = "clusters"
+    FARGATE_PROFILE_NAMES: str = "fargateProfileNames"
+    FARGATE_PROFILE: str = "fargateProfile"
+    NEXT_TOKEN: str = "nextToken"
+    NODEGROUP: str = "nodegroup"
+    NODEGROUPS: str = "nodegroups"
 
 
 class ErrorAttributes:
     """Key names for the dictionaries representing error messages."""
 
-    CODE: slice = "Code"
-    ERROR: slice = "Error"
-    MESSAGE: slice = "Message"
+    CODE: str = "Code"
+    ERROR: str = "Error"
+    MESSAGE: str = "Message"
 
 
 class ClusterInputs:
@@ -137,37 +137,37 @@ class PossibleTestResults(Enum):
 class ClusterAttributes:
     """Key names for the dictionaries representing EKS Clusters."""
 
-    ARN: slice = "arn"
-    CLUSTER_NAME: slice = "clusterName"
-    CREATED_AT: slice = "createdAt"
-    ENDPOINT: slice = "endpoint"
-    IDENTITY: slice = "identity"
-    ISSUER: slice = "issuer"
-    NAME: slice = "name"
-    OIDC: slice = "oidc"
+    ARN: str = "arn"
+    CLUSTER_NAME: str = "clusterName"
+    CREATED_AT: str = "createdAt"
+    ENDPOINT: str = "endpoint"
+    IDENTITY: str = "identity"
+    ISSUER: str = "issuer"
+    NAME: str = "name"
+    OIDC: str = "oidc"
 
 
 class FargateProfileAttributes:
-    ARN: slice = "fargateProfileArn"
-    CREATED_AT: slice = "createdAt"
-    FARGATE_PROFILE_NAME: slice = "fargateProfileName"
-    LABELS: slice = "labels"
-    NAMESPACE: slice = "namespace"
-    SELECTORS: slice = "selectors"
+    ARN: str = "fargateProfileArn"
+    CREATED_AT: str = "createdAt"
+    FARGATE_PROFILE_NAME: str = "fargateProfileName"
+    LABELS: str = "labels"
+    NAMESPACE: str = "namespace"
+    SELECTORS: str = "selectors"
 
 
 class NodegroupAttributes:
     """Key names for the dictionaries representing EKS Managed Nodegroups."""
 
-    ARN: slice = "nodegroupArn"
-    AUTOSCALING_GROUPS: slice = "autoScalingGroups"
-    CREATED_AT: slice = "createdAt"
-    MODIFIED_AT: slice = "modifiedAt"
-    NAME: slice = "name"
-    NODEGROUP_NAME: slice = "nodegroupName"
-    REMOTE_ACCESS_SG: slice = "remoteAccessSecurityGroup"
-    RESOURCES: slice = "resources"
-    TAGS: slice = "tags"
+    ARN: str = "nodegroupArn"
+    AUTOSCALING_GROUPS: str = "autoScalingGroups"
+    CREATED_AT: str = "createdAt"
+    MODIFIED_AT: str = "modifiedAt"
+    NAME: str = "name"
+    NODEGROUP_NAME: str = "nodegroupName"
+    REMOTE_ACCESS_SG: str = "remoteAccessSecurityGroup"
+    RESOURCES: str = "resources"
+    TAGS: str = "tags"
 
 
 class BatchCountSize:
diff --git a/tests/providers/elasticsearch/log/elasticmock/__init__.py b/tests/providers/elasticsearch/log/elasticmock/__init__.py
index e60eef3d230a0..18008d00a9458 100644
--- a/tests/providers/elasticsearch/log/elasticmock/__init__.py
+++ b/tests/providers/elasticsearch/log/elasticmock/__init__.py
@@ -42,7 +42,7 @@
 from typing import Dict
 from unittest.mock import patch
 
-from elasticsearch.client import _normalize_hosts
+from elasticsearch.client.utils import _normalize_hosts  # type: ignore
 
 from .fake_elasticsearch import FakeElasticsearch
 
diff --git a/tests/providers/microsoft/azure/hooks/test_azure_data_factory.py b/tests/providers/microsoft/azure/hooks/test_azure_data_factory.py
index 60e7522f60ee9..d4c1d95321d66 100644
--- a/tests/providers/microsoft/azure/hooks/test_azure_data_factory.py
+++ b/tests/providers/microsoft/azure/hooks/test_azure_data_factory.py
@@ -177,7 +177,7 @@ def test_create_factory(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((MODEL,), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, MODEL)),
 )
 def test_update_factory(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._factory_exists = Mock(return_value=True)
+    hook._factory_exists = Mock(return_value=True)  # type: ignore
     hook.update_factory(*user_args)
 
     hook._conn.factories.create_or_update.assert_called_with(*sdk_args)
@@ -188,7 +188,7 @@ def test_update_factory(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((MODEL,), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, MODEL)),
 )
 def test_update_factory_non_existent(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._factory_exists = Mock(return_value=False)
+    hook._factory_exists = Mock(return_value=False)  # type: ignore
 
     with pytest.raises(AirflowException, match=r"Factory .+ does not exist"):
         hook.update_factory(*user_args)
@@ -229,7 +229,7 @@ def test_create_linked_service(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
 )
 def test_update_linked_service(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._linked_service_exists = Mock(return_value=True)
+    hook._linked_service_exists = Mock(return_value=True)  # type: ignore
     hook.update_linked_service(*user_args)
 
     hook._conn.linked_services.create_or_update(*sdk_args)
@@ -240,7 +240,7 @@ def test_update_linked_service(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
 )
 def test_update_linked_service_non_existent(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._linked_service_exists = Mock(return_value=False)
+    hook._linked_service_exists = Mock(return_value=False)  # type: ignore
 
     with pytest.raises(AirflowException, match=r"Linked service .+ does not exist"):
         hook.update_linked_service(*user_args)
@@ -281,7 +281,7 @@ def test_create_dataset(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
 )
 def test_update_dataset(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._dataset_exists = Mock(return_value=True)
+    hook._dataset_exists = Mock(return_value=True)  # type: ignore
     hook.update_dataset(*user_args)
 
     hook._conn.datasets.create_or_update.assert_called_with(*sdk_args)
@@ -292,7 +292,7 @@ def test_update_dataset(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
 )
 def test_update_dataset_non_existent(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._dataset_exists = Mock(return_value=False)
+    hook._dataset_exists = Mock(return_value=False)  # type: ignore
 
     with pytest.raises(AirflowException, match=r"Dataset .+ does not exist"):
         hook.update_dataset(*user_args)
@@ -333,7 +333,7 @@ def test_create_pipeline(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
 )
 def test_update_pipeline(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._pipeline_exists = Mock(return_value=True)
+    hook._pipeline_exists = Mock(return_value=True)  # type: ignore
     hook.update_pipeline(*user_args)
 
     hook._conn.pipelines.create_or_update.assert_called_with(*sdk_args)
@@ -344,7 +344,7 @@ def test_update_pipeline(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
 )
 def test_update_pipeline_non_existent(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._pipeline_exists = Mock(return_value=False)
+    hook._pipeline_exists = Mock(return_value=False)  # type: ignore
 
     with pytest.raises(AirflowException, match=r"Pipeline .+ does not exist"):
         hook.update_pipeline(*user_args)
@@ -451,7 +451,7 @@ def test_create_trigger(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
 )
 def test_update_trigger(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._trigger_exists = Mock(return_value=True)
+    hook._trigger_exists = Mock(return_value=True)  # type: ignore
     hook.update_trigger(*user_args)
 
     hook._conn.triggers.create_or_update.assert_called_with(*sdk_args)
@@ -462,7 +462,7 @@ def test_update_trigger(hook: AzureDataFactoryHook, user_args, sdk_args):
     implicit_factory=((NAME, MODEL), (DEFAULT_RESOURCE_GROUP, DEFAULT_FACTORY, NAME, MODEL)),
 )
 def test_update_trigger_non_existent(hook: AzureDataFactoryHook, user_args, sdk_args):
-    hook._trigger_exists = Mock(return_value=False)
+    hook._trigger_exists = Mock(return_value=False)  # type: ignore
 
     with pytest.raises(AirflowException, match=r"Trigger .+ does not exist"):
         hook.update_trigger(*user_args)
diff --git a/tests/providers/mongo/hooks/test_mongo.py b/tests/providers/mongo/hooks/test_mongo.py
index d52a3b53f5bed..d6f2d7748ae2a 100644
--- a/tests/providers/mongo/hooks/test_mongo.py
+++ b/tests/providers/mongo/hooks/test_mongo.py
@@ -26,7 +26,7 @@
 try:
     import mongomock
 except ImportError:
-    mongomock = None
+    mongomock = None  # type: ignore
 
 
 class MongoHookTest(MongoHook):
diff --git a/tests/providers/sftp/sensors/test_sftp.py b/tests/providers/sftp/sensors/test_sftp.py
index 682e6eb83e97e..5bccfa9be1197 100644
--- a/tests/providers/sftp/sensors/test_sftp.py
+++ b/tests/providers/sftp/sensors/test_sftp.py
@@ -20,7 +20,7 @@
 from unittest.mock import patch
 
 import pytest
-from paramiko import SFTP_FAILURE, SFTP_NO_SUCH_FILE
+from paramiko.sftp import SFTP_FAILURE, SFTP_NO_SUCH_FILE
 
 from airflow.providers.sftp.sensors.sftp import SFTPSensor
 
diff --git a/tests/test_utils/mock_security_manager.py b/tests/test_utils/mock_security_manager.py
index 4c0e3fec687d9..f91482441f93f 100644
--- a/tests/test_utils/mock_security_manager.py
+++ b/tests/test_utils/mock_security_manager.py
@@ -15,6 +15,10 @@
 # specific language governing permissions and limitations
 # under the License.
 
+# Ignore type checking here, the FAB security manager has been ported from FAB
+# and it has a bit too many class variables/dynamic typing to bother
+# type: ignore
+
 from airflow.www.security import AirflowSecurityManager
 
 
diff --git a/tests/utils/test_email.py b/tests/utils/test_email.py
index c7f228a21e759..fb0e14da908ce 100644
--- a/tests/utils/test_email.py
+++ b/tests/utils/test_email.py
@@ -273,7 +273,7 @@ def test_send_mime_dryrun(self, mock_smtp, mock_smtp_ssl):
 
     @mock.patch('smtplib.SMTP_SSL')
     @mock.patch('smtplib.SMTP')
-    def test_send_mime_complete_failure(self, mock_smtp: mock, mock_smtp_ssl):
+    def test_send_mime_complete_failure(self, mock_smtp, mock_smtp_ssl):
         mock_smtp.side_effect = SMTPServerDisconnected()
         msg = MIMEMultipart()
         with pytest.raises(SMTPServerDisconnected):
