Migrate to connexion 2.15

This PR migrates Airflow to connexion 2.15 - which allows Airflow
to bump a number of dependencies such as Flask, Werkzeug, flask-session.

There are a few incompatibilities fixed during the migration introduced
by breaking changes in those dependencies:

* flask-session >= 0.8.0 manages it's own session table and we already
  managed it on our own, so we conditionally create/drop the table when
  running migrations.

* flask-session >= 0.7.9 switched to use MsgSpec as default serializer
  and while it claims you can switch back to TagJSONSerializer, it's not
  workign out of the box because it relies on encode/decode methods of
  the serializer that are not present in TagJSONSerializer. But we
  cannot use MsgSpec serializer because it does not support some objects
  (Markup) that we use for Flash message serialization. We borrow a hack
  from ckan/ckan#8704 and provide our custom
  serializer deriving from TagJSONSerializer with encode/decode methods
  added

* flask-session >= 0.8.0 switched to use "client" instead of "db" as
  interface field. Our code was adapted accordingly

* Werkzeug >= 2.3.0 changed the way how percent-encoding is done -
  instead of percent-encoding every reserved character, only the
  reserved characters that have "reserved purpose" in the place of
  the URL that they are used are encoded - following WhatWG
  specification. This means for example that "&" is not encoded in
  path, "/:" are not encoded in url params, and "?" is not encoded
  after it is used to separate url from parameters. Our tests expected
  percent-encoding everywherei (previous Werkzeug behaviour), so they
  had to be updated.

* Flask test client in newer versions of flask does not have "cookie_jar"
  - it exposes cookies via "get_cookie", tests were updated to use it.

* Minimum versions of dependencies have been updated so that all the
  changes are compatible with them and we do not have to handle
  back-compatibility with older versions.

Author: potiuk

airflow/__init__.py

@@ -17,7 +17,7 @@
 # under the License.
 from __future__ import annotations
 
-__version__ = "2.11.0"
+__version__ = "2.11.1"
 
 import os
 import sys

airflow/api_internal/internal_api_call.py

@@ -135,7 +135,7 @@ def _is_retryable_exception(exception: BaseException) -> bool:
         stop=tenacity.stop_after_attempt(10),
         wait=tenacity.wait_exponential(min=1),
         retry=tenacity.retry_if_exception(_is_retryable_exception),
-        before_sleep=tenacity.before_log(logger, logging.WARNING),
+        before_sleep=tenacity.before_log(logger, logging.WARNING),  # type: ignore[arg-type]
     )
     def make_jsonrpc_request(method_name: str, params_json: str) -> bytes:
         signer = JWTSigner(

airflow/cli/commands/info_command.py

@@ -352,8 +352,8 @@ class FileIoException(Exception):
     stop=tenacity.stop_after_attempt(5),
     wait=tenacity.wait_exponential(multiplier=1, max=10),
     retry=tenacity.retry_if_exception_type(FileIoException),
-    before=tenacity.before_log(log, logging.DEBUG),
-    after=tenacity.after_log(log, logging.DEBUG),
+    before=tenacity.before_log(log, logging.DEBUG),  # type: ignore[arg-type]
+    after=tenacity.after_log(log, logging.DEBUG),  # type: ignore[arg-type]
 )
 def _upload_text_to_fileio(content):
     """Upload text file to File.io service and return link."""

airflow/example_dags/example_branch_operator.py

@@ -128,7 +128,7 @@ def hello_world_with_external_python():
 
         # [START howto_operator_branch_virtualenv]
         # Note: Passing a caching dir allows to keep the virtual environment over multiple runs
-        #       Run the example a second time and see that it re-uses it and is faster.
+        #       Run the example a second time and see that it reuses it and is faster.
         VENV_CACHE_PATH = Path(tempfile.gettempdir())
 
         def branch_with_venv(choices):

airflow/example_dags/example_branch_operator_decorator.py

@@ -114,7 +114,7 @@ def some_ext_py_task():
 
         # [START howto_operator_branch_virtualenv]
         # Note: Passing a caching dir allows to keep the virtual environment over multiple runs
-        #       Run the example a second time and see that it re-uses it and is faster.
+        #       Run the example a second time and see that it reuses it and is faster.
         VENV_CACHE_PATH = tempfile.gettempdir()
 
         @task.branch_virtualenv(requirements=["numpy~=1.26.0"], venv_cache_path=VENV_CACHE_PATH)

airflow/kubernetes/pre_7_4_0_compatibility/kube_client.py

@@ -86,8 +86,8 @@ def _enable_tcp_keepalive() -> None:
     else:
         log.debug("Unable to set TCP_KEEPCNT on this platform")
 
-    HTTPSConnection.default_socket_options = HTTPSConnection.default_socket_options + socket_options
-    HTTPConnection.default_socket_options = HTTPConnection.default_socket_options + socket_options
+    HTTPSConnection.default_socket_options = HTTPSConnection.default_socket_options + socket_options  # type: ignore[operator]
+    HTTPConnection.default_socket_options = HTTPConnection.default_socket_options + socket_options  # type: ignore[operator]
 
 
 def get_kube_client(

airflow/migrations/versions/0095_2_2_4_add_session_table_to_db.py

@@ -49,9 +49,10 @@ def upgrade():
         sa.Column("expiry", sa.DateTime()),
         sa.PrimaryKeyConstraint("id"),
         sa.UniqueConstraint("session_id"),
+        if_not_exists=True,
     )
 
 
 def downgrade():
     """Unapply Create a ``session`` table to store web session data."""
-    op.drop_table(TABLE_NAME)
+    op.drop_table(TABLE_NAME, if_exists=True)

airflow/models/dag.py

@@ -2014,7 +2014,7 @@ def _get_task_instances(
                 visited_external_tis = set()
 
             p_dag = self.parent_dag.partial_subset(
-                task_ids_or_regex=r"^{}$".format(self.dag_id.split(".")[1]),
+                task_ids_or_regex=rf"^{self.dag_id.split('.')[1]}$",
                 include_upstream=False,
                 include_downstream=True,
             )

airflow/operators/python.py

@@ -820,7 +820,7 @@ def _ensure_venv_cache_exists(self, venv_cache_path: Path) -> Path:
                     if hash_marker.exists():
                         previous_hash_data = hash_marker.read_text(encoding="utf8")
                         if previous_hash_data == hash_data:
-                            self.log.info("Re-using cached Python virtual environment in %s", venv_path)
+                            self.log.info("Reusing cached Python virtual environment in %s", venv_path)
                             return venv_path
 
                         _, hash_data_before_upgrade = self._calculate_cache_hash(exclude_cloudpickle=True)

airflow/providers/alibaba/cloud/log/oss_task_handler.py

@@ -72,7 +72,7 @@ def set_context(self, ti):
         self.upload_on_close = not ti.raw
 
         # Clear the file first so that duplicate data is not uploaded
-        # when re-using the same path (e.g. with rescheduled sensors)
+        # when reusing the same path (e.g. with rescheduled sensors)
         if self.upload_on_close:
             with open(self.handler.baseFilename, "w"):
                 pass