Remove airflow.www module

airflow/dag_processing/collection.py

@@ -217,7 +217,7 @@ def _sync_dag_perms(dag: MaybeSerializedDAG, session: Session):
     dag_id = dag.dag_id
 
     log.debug("Syncing DAG permissions: %s to the DB", dag_id)
-    from airflow.www.security_appless import ApplessAirflowSecurityManager
+    from airflow.providers.fab.www.security_appless import ApplessAirflowSecurityManager
 
     security_manager = ApplessAirflowSecurityManager(session=session)
     security_manager.sync_perm_for_dag(dag_id, dag.access_control)

airflow/utils/db.py

@@ -735,7 +735,7 @@ def _get_flask_db(sql_database_uri):
     from flask import Flask
     from flask_sqlalchemy import SQLAlchemy
 
-    from airflow.www.session import AirflowDatabaseSessionInterface
+    from airflow.providers.fab.www.session import AirflowDatabaseSessionInterface
 
     flask_app = Flask(__name__)
     flask_app.config["SQLALCHEMY_DATABASE_URI"] = sql_database_uri

providers/edge/src/airflow/providers/edge/plugins/edge_executor_plugin.py

@@ -24,6 +24,7 @@
 
 from flask import Blueprint, redirect, request, url_for
 from flask_appbuilder import BaseView, expose
+from markupsafe import Markup
 from sqlalchemy import select
 
 from airflow.auth.managers.models.resource_details import AccessView
@@ -32,14 +33,14 @@
 from airflow.models.taskinstance import TaskInstanceState
 from airflow.plugins_manager import AirflowPlugin
 from airflow.providers.edge.version_compat import AIRFLOW_V_3_0_PLUS
+from airflow.utils.state import State
 
 if AIRFLOW_V_3_0_PLUS:
     from airflow.providers.fab.www.auth import has_access_view
 else:
     from airflow.www.auth import has_access_view  # type: ignore
 from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.utils.yaml import safe_load
-from airflow.www import utils as wwwutils
 
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
@@ -80,6 +81,18 @@ def _get_api_endpoint() -> dict[str, Any]:
     }
 
 
+def _state_token(state):
+    """Return a formatted string with HTML for a given State."""
+    color = State.color(state)
+    fg_color = State.color_fg(state)
+    return Markup(
+        """
+        <span class="label" style="color:{fg_color}; background-color:{color};"
+            title="Current State: {state}">{state}</span>
+        """
+    ).format(color=color, state=state, fg_color=fg_color)
+
+
 def modify_maintenance_comment_on_update(maintenance_comment: str | None, username: str) -> str:
     if maintenance_comment:
         if re.search(
@@ -121,7 +134,7 @@ def jobs(self, session: Session = NEW_SESSION):
 
         jobs = session.scalars(select(EdgeJobModel).order_by(EdgeJobModel.queued_dttm)).all()
         html_states = {
-            str(state): wwwutils.state_token(str(state)) for state in TaskInstanceState.__members__.values()
+            str(state): _state_token(str(state)) for state in TaskInstanceState.__members__.values()
         }
         return self.render_template("edge_worker_jobs.html", jobs=jobs, html_states=html_states)
 

providers/fab/tests/unit/fab/utils.py

@@ -211,7 +211,6 @@ def local_context(self):
             # flask_appbuilder.baseviews.BaseView.render_template
             "appbuilder",
             "base_template",
-            # airflow.www.app.py.create_app (inner method - jinja_globals)
             "server_timezone",
             "default_ui_timezone",
             "hostname",
@@ -227,11 +226,8 @@ def local_context(self):
             "airflow_version",
             "git_version",
             "k8s_or_k8scelery_executor",
-            # airflow.www.static_config.configure_manifest_files
             "url_for_asset",
-            # airflow.www.views.AirflowBaseView.render_template
             "scheduler_job",
-            # airflow.www.views.AirflowBaseView.extra_args
             "macros",
             "auth_manager",
             "triggerer_job",

providers/google/src/airflow/providers/google/cloud/hooks/bigquery.py

@@ -120,7 +120,7 @@ def get_connection_form_widgets(cls) -> dict[str, Any]:
         from wtforms import validators
         from wtforms.fields.simple import BooleanField, StringField
 
-        from airflow.www.validators import ValidJson
+        from airflow.providers.google.cloud.utils.validators import ValidJson
 
         connection_form_widgets = super().get_connection_form_widgets()
         connection_form_widgets["use_legacy_sql"] = BooleanField(lazy_gettext("Use Legacy SQL"), default=True)

airflow/www/utils.py → providers/google/src/airflow/providers/google/cloud/utils/validators.py

@@ -17,18 +17,27 @@
 # under the License.
 from __future__ import annotations
 
-from markupsafe import Markup
+import json
+from json import JSONDecodeError
 
-from airflow.utils.state import State
+from wtforms.validators import ValidationError
 
 
-def state_token(state):
-    """Return a formatted string with HTML for a given State."""
-    color = State.color(state)
-    fg_color = State.color_fg(state)
-    return Markup(
-        """
-        <span class="label" style="color:{fg_color}; background-color:{color};"
-            title="Current State: {state}">{state}</span>
-        """
-    ).format(color=color, state=state, fg_color=fg_color)
+class ValidJson:
+    """
+    Validates data is valid JSON.
+
+    :param message:
+        Error message to raise in case of a validation error.
+    """
+
+    def __init__(self, message=None):
+        self.message = message
+
+    def __call__(self, form, field):
+        if field.data:
+            try:
+                json.loads(field.data)
+            except JSONDecodeError as ex:
+                message = self.message or f"JSON Validation Error: {ex}"
+                raise ValidationError(message=field.gettext(message.format(field.data)))

providers/google/tests/unit/google/cloud/utils/test_validators.py

@@ -0,0 +1,59 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+from wtforms.validators import ValidationError
+
+from airflow.providers.google.cloud.utils.validators import ValidJson
+
+
+class TestValidJson:
+    def setup_method(self):
+        self.form_field_mock = mock.MagicMock(data='{"valid":"True"}')
+        self.form_field_mock.gettext.side_effect = lambda msg: msg
+        self.form_mock = mock.MagicMock(spec_set=dict)
+
+    def _validate(self, message=None):
+        validator = ValidJson(message=message)
+
+        return validator(self.form_mock, self.form_field_mock)
+
+    def test_form_field_is_none(self):
+        self.form_field_mock.data = None
+
+        assert self._validate() is None
+
+    def test_validation_pass(self):
+        assert self._validate() is None
+
+    def test_validation_raises_default_message(self):
+        self.form_field_mock.data = "2017-05-04"
+
+        with pytest.raises(ValidationError, match="JSON Validation Error:.*"):
+            self._validate()
+
+    def test_validation_raises_custom_message(self):
+        self.form_field_mock.data = "2017-05-04"
+
+        with pytest.raises(ValidationError, match="Invalid JSON"):
+            self._validate(
+                message="Invalid JSON: {}",
+            )