[DRAFT] Switch to Connexion 3 framework

.github/workflows/basic-tests.yml

@@ -148,6 +148,8 @@ jobs:
         env:
           HATCH_ENV: "test"
         working-directory: ./clients/python
+      - name: Compile www assets
+        run: breeze compile-www-assets
       - name: "Install Airflow in editable mode with fab for webserver tests"
         run: pip install -e ".[fab]"
       - name: "Install Python client"

airflow/api_connexion/endpoints/connection_endpoint.py

@@ -91,7 +91,7 @@ def get_connection(*, connection_id: str, session: Session = NEW_SESSION) -> API
 @provide_session
 def get_connections(
     *,
-    limit: int,
+    limit: int | None = None,
     offset: int = 0,
     order_by: str = "id",
     session: Session = NEW_SESSION,

airflow/api_connexion/endpoints/dag_endpoint.py

@@ -94,7 +94,7 @@ def get_dag_details(
 @provide_session
 def get_dags(
     *,
-    limit: int,
+    limit: int | None = None,
     offset: int = 0,
     tags: Collection[str] | None = None,
     dag_id_pattern: str | None = None,

airflow/api_connexion/endpoints/dag_parsing.py

@@ -19,7 +19,8 @@
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Sequence
 
-from flask import Response, current_app
+from connexion import NoContent
+from flask import current_app
 from itsdangerous import BadSignature, URLSafeSerializer
 from sqlalchemy import exc, select
 
@@ -39,7 +40,9 @@
 
 @security.requires_access_dag("PUT")
 @provide_session
-def reparse_dag_file(*, file_token: str, session: Session = NEW_SESSION) -> Response:
+def reparse_dag_file(
+    *, file_token: str, session: Session = NEW_SESSION
+) -> tuple[str | NoContent, HTTPStatus]:
     """Request re-parsing a DAG file."""
     secret_key = current_app.config["SECRET_KEY"]
     auth_s = URLSafeSerializer(secret_key)
@@ -65,5 +68,5 @@ def reparse_dag_file(*, file_token: str, session: Session = NEW_SESSION) -> Resp
         session.commit()
     except exc.IntegrityError:
         session.rollback()
-        return Response("Duplicate request", HTTPStatus.CREATED)
-    return Response(status=HTTPStatus.CREATED)
+        return "Duplicate request", HTTPStatus.CREATED
+    return NoContent, HTTPStatus.CREATED

airflow/api_connexion/endpoints/dag_warning_endpoint.py

@@ -43,7 +43,7 @@
 @provide_session
 def get_dag_warnings(
     *,
-    limit: int,
+    limit: int | None = None,
     dag_id: str | None = None,
     warning_type: str | None = None,
     offset: int | None = None,

airflow/api_connexion/endpoints/dataset_endpoint.py

@@ -82,7 +82,7 @@ def get_dataset(*, uri: str, session: Session = NEW_SESSION) -> APIResponse:
 @provide_session
 def get_datasets(
     *,
-    limit: int,
+    limit: int | None = None,
     offset: int = 0,
     uri_pattern: str | None = None,
     dag_ids: str | None = None,
@@ -113,11 +113,11 @@ def get_datasets(
 
 
 @security.requires_access_dataset("GET")
-@provide_session
 @format_parameters({"limit": check_limit})
+@provide_session
 def get_dataset_events(
     *,
-    limit: int,
+    limit: int | None = None,
     offset: int = 0,
     order_by: str = "timestamp",
     dataset_id: int | None = None,

airflow/api_connexion/endpoints/event_log_endpoint.py

@@ -64,7 +64,7 @@ def get_event_logs(
     included_events: str | None = None,
     before: str | None = None,
     after: str | None = None,
-    limit: int,
+    limit: int | None = None,
     offset: int | None = None,
     order_by: str = "event_log_id",
     session: Session = NEW_SESSION,

airflow/api_connexion/endpoints/import_error_endpoint.py

@@ -77,7 +77,7 @@ def get_import_error(*, import_error_id: int, session: Session = NEW_SESSION) ->
 @provide_session
 def get_import_errors(
     *,
-    limit: int,
+    limit: int | None = None,
     offset: int | None = None,
     order_by: str = "import_error_id",
     session: Session = NEW_SESSION,

airflow/api_connexion/endpoints/log_endpoint.py

@@ -107,7 +107,10 @@ def get_log(
         logs = logs[0] if task_try_number is not None else logs
         # we must have token here, so we can safely ignore it
         token = URLSafeSerializer(key).dumps(metadata)  # type: ignore[assignment]
-        return logs_schema.dump(LogResponseObject(continuation_token=token, content=logs))
+        return Response(
+            logs_schema.dumps(LogResponseObject(continuation_token=token, content=logs)),
+            headers={"Content-Type": "application/json"},
+        )
     # text/plain. Stream
     logs = task_log_reader.read_log_stream(ti, task_try_number, metadata)
 

airflow/api_connexion/endpoints/pool_endpoint.py

@@ -68,7 +68,7 @@ def get_pool(*, pool_name: str, session: Session = NEW_SESSION) -> APIResponse:
 @provide_session
 def get_pools(
     *,
-    limit: int,
+    limit: int | None = None,
     order_by: str = "id",
     offset: int | None = None,
     session: Session = NEW_SESSION,

airflow/api_connexion/endpoints/task_instance_endpoint.py

@@ -309,7 +309,7 @@ def _apply_range_filter(query: Select, key: ClauseElement, value_range: tuple[T,
 @provide_session
 def get_task_instances(
     *,
-    limit: int,
+    limit: int | None = None,
     dag_id: str | None = None,
     dag_run_id: str | None = None,
     execution_date_gte: str | None = None,

airflow/api_connexion/exceptions.py

@@ -19,13 +19,12 @@
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Any
 
-import werkzeug
-from connexion import FlaskApi, ProblemException, problem
+from connexion import ProblemException, problem
 
 from airflow.utils.docs import get_docs_url
 
 if TYPE_CHECKING:
-    import flask
+    from connexion.lifecycle import ConnexionRequest, ConnexionResponse
 
 doc_link = get_docs_url("stable-rest-api-ref.html")
 
@@ -40,37 +39,29 @@
 }
 
 
-def common_error_handler(exception: BaseException) -> flask.Response:
+def problem_error_handler(_request: ConnexionRequest, exception: ProblemException) -> ConnexionResponse:
     """Use to capture connexion exceptions and add link to the type field."""
-    if isinstance(exception, ProblemException):
-        link = EXCEPTIONS_LINK_MAP.get(exception.status)
-        if link:
-            response = problem(
-                status=exception.status,
-                title=exception.title,
-                detail=exception.detail,
-                type=link,
-                instance=exception.instance,
-                headers=exception.headers,
-                ext=exception.ext,
-            )
-        else:
-            response = problem(
-                status=exception.status,
-                title=exception.title,
-                detail=exception.detail,
-                type=exception.type,
-                instance=exception.instance,
-                headers=exception.headers,
-                ext=exception.ext,
-            )
+    link = EXCEPTIONS_LINK_MAP.get(exception.status)
+    if link:
+        return problem(
+            status=exception.status,
+            title=exception.title,
+            detail=exception.detail,
+            type=link,
+            instance=exception.instance,
+            headers=exception.headers,
+            ext=exception.ext,
+        )
     else:
-        if not isinstance(exception, werkzeug.exceptions.HTTPException):
-            exception = werkzeug.exceptions.InternalServerError()
-
-        response = problem(title=exception.name, detail=exception.description, status=exception.code)
-
-    return FlaskApi.get_response(response)
+        return problem(
+            status=exception.status,
+            title=exception.title,
+            detail=exception.detail,
+            type=exception.type,
+            instance=exception.instance,
+            headers=exception.headers,
+            ext=exception.ext,
+        )
 
 
 class NotFound(ProblemException):

airflow/api_connexion/openapi/v1.yaml

@@ -1453,6 +1453,10 @@ paths:
       responses:
         "204":
           description: Success.
+          content:
+            text/html:
+               schema:
+                 type: string
         "400":
           $ref: "#/components/responses/BadRequest"
         "401":
@@ -1831,6 +1835,10 @@ paths:
       responses:
         "204":
           description: Success.
+          content:
+            text/html:
+               schema:
+                 type: string
         "400":
           $ref: "#/components/responses/BadRequest"
         "401":
@@ -1973,8 +1981,8 @@ paths:
         response = self.client.get(
             request_url,
             query_string={"token": token},
-            headers={"Accept": "text/plain"},
-            environ_overrides={"REMOTE_USER": "test"},
+            headers={"Accept": "text/plain","REMOTE_USER": "test"},
+
         )
         continuation_token = response.json["continuation_token"]
             metadata = URLSafeSerializer(key).loads(continuation_token)
@@ -2108,7 +2116,7 @@ paths:
                 properties:
                   content:
                     type: string
-            plain/text:
+            text/plain:
               schema:
                 type: string
 
@@ -2194,29 +2202,6 @@ paths:
         "403":
           $ref: "#/components/responses/PermissionDenied"
 
-  /datasets/{uri}:
-    parameters:
-      - $ref: "#/components/parameters/DatasetURI"
-    get:
-      summary: Get a dataset
-      description: Get a dataset by uri.
-      x-openapi-router-controller: airflow.api_connexion.endpoints.dataset_endpoint
-      operationId: get_dataset
-      tags: [Dataset]
-      responses:
-        "200":
-          description: Success.
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/Dataset"
-        "401":
-          $ref: "#/components/responses/Unauthenticated"
-        "403":
-          $ref: "#/components/responses/PermissionDenied"
-        "404":
-          $ref: "#/components/responses/NotFound"
-
   /datasets/events:
     get:
       summary: Get dataset events
@@ -2274,6 +2259,30 @@ paths:
         '404':
           $ref: '#/components/responses/NotFound'
 
+  /datasets/{uri}:
+    parameters:
+      - $ref: "#/components/parameters/DatasetURI"
+    get:
+      summary: Get a dataset
+      description: Get a dataset by uri.
+      x-openapi-router-controller: airflow.api_connexion.endpoints.dataset_endpoint
+      operationId: get_dataset
+      tags: [Dataset]
+      responses:
+        "200":
+          description: Success.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Dataset"
+        "401":
+          $ref: "#/components/responses/Unauthenticated"
+        "403":
+          $ref: "#/components/responses/PermissionDenied"
+        "404":
+          $ref: "#/components/responses/NotFound"
+
+
   /config:
     get:
       summary: Get current configuration

airflow/api_connexion/parameters.py

@@ -41,7 +41,7 @@ def validate_istimezone(value: datetime) -> None:
         raise BadRequest("Invalid datetime format", detail="Naive datetime is disallowed")
 
 
-def format_datetime(value: str) -> datetime:
+def format_datetime(value: str | None) -> datetime | None:
     """
     Format datetime objects.
 
@@ -50,6 +50,8 @@ def format_datetime(value: str) -> datetime:
 
     This should only be used within connection views because it raises 400
     """
+    if value is None:
+        return None
     value = value.strip()
     if value[-1] != "Z":
         value = value.replace(" ", "+")
@@ -59,7 +61,7 @@ def format_datetime(value: str) -> datetime:
         raise BadRequest("Incorrect datetime argument", detail=str(err))
 
 
-def check_limit(value: int) -> int:
+def check_limit(value: int | None) -> int:
     """
     Check the limit does not exceed configured value.
 
@@ -68,7 +70,8 @@ def check_limit(value: int) -> int:
     """
     max_val = conf.getint("api", "maximum_page_limit")  # user configured max page limit
     fallback = conf.getint("api", "fallback_page_limit")
-
+    if value is None:
+        return fallback
     if value > max_val:
         log.warning(
             "The limit param value %s passed in API exceeds the configured maximum page limit %s",
@@ -99,8 +102,9 @@ def format_parameters_decorator(func: T) -> T:
         @wraps(func)
         def wrapped_function(*args, **kwargs):
             for key, formatter in params_formatters.items():
-                if key in kwargs:
-                    kwargs[key] = formatter(kwargs[key])
+                value = formatter(kwargs.get(key))
+                if value:
+                    kwargs[key] = value
             return func(*args, **kwargs)
 
         return cast(T, wrapped_function)

airflow/auth/managers/base_auth_manager.py

@@ -19,7 +19,7 @@
 
 from abc import abstractmethod
 from functools import cached_property
-from typing import TYPE_CHECKING, Container, Literal, Sequence
+from typing import TYPE_CHECKING, Any, Container, Literal, Sequence
 
 from flask_appbuilder.menu import MenuItem
 from sqlalchemy import select
@@ -82,7 +82,7 @@ def get_cli_commands() -> list[CLICommand]:
         return []
 
     def get_api_endpoints(self) -> None | Blueprint:
-        """Return API endpoint(s) definition for the auth manager."""
+        """Return API endpoint(s) definition for the auth manager for Airflow 2.9."""
         return None
 
     def get_user_name(self) -> str:
@@ -442,3 +442,12 @@ def security_manager(self) -> AirflowSecurityManagerV2:
         from airflow.www.security_manager import AirflowSecurityManagerV2
 
         return AirflowSecurityManagerV2(self.appbuilder)
+
+    def get_auth_manager_api_specification(self) -> tuple[str | None, dict[Any, Any]]:
+        """
+        Return the mount point and specification (openapi) for auth manager contributed API (Airflow 2.10).
+
+        By default is raises NotImplementedError which produces a warning in airflow logs when auth manager is
+        initialized, but you can return None, {} if the auth manager does not contribute API.
+        """
+        raise NotImplementedError