fixup! Composer webserver patch

Cherry-pick apache/airflow#41350

Change-Id: If9d27e06c3f9a686b5d9869986c6a92213d6b83a
GitOrigin-RevId: 70abf450fd69e59ef79f7ed9addde9900761e229

Author: Cloud Composer Team

airflow/www/views.py

@@ -2023,7 +2023,7 @@ def trigger(self, dag_id: str, session: Session = NEW_SESSION):
             .limit(num_recent_confs)
         )
         recent_confs = {
-            run_id: json.dumps(run_conf)
+            run_id: json.dumps(run_conf, cls=utils_json.WebEncoder)
             for run_id, run_conf in ((run.run_id, run.conf) for run in recent_runs)
             if isinstance(run_conf, dict) and any(run_conf)
         }
@@ -2058,6 +2058,7 @@ def trigger(self, dag_id: str, session: Session = NEW_SESSION):
                         },
                         indent=4,
                         ensure_ascii=False,
+                        cls=utils_json.WebEncoder,
                     )
                 except TypeError:
                     flash("Could not pre-populate conf field due to non-JSON-serializable data-types")

tests/www/views/test_views_trigger_dag.py

@@ -19,6 +19,7 @@
 
 import datetime
 import json
+from decimal import Decimal
 from urllib.parse import quote
 
 import pytest
@@ -28,6 +29,7 @@
 from airflow.operators.empty import EmptyOperator
 from airflow.security import permissions
 from airflow.utils import timezone
+from airflow.utils.json import WebEncoder
 from airflow.utils.session import create_session
 from airflow.utils.types import DagRunType
 from tests.test_utils.api_connexion_utils import create_test_client
@@ -92,6 +94,32 @@ def test_trigger_dag_conf(admin_client):
     assert run.conf == conf_dict
 
 
+def test_trigger_dag_conf_serializable_fields(admin_client):
+    test_dag_id = "example_bash_operator"
+    time_now = timezone.utcnow()
+    conf_dict = {
+        "string": "Hello, World!",
+        "date_str": "2024-08-08T09:57:35.300858",
+        "datetime": time_now,
+        "decimal": Decimal(10.465),
+    }
+    expected_conf = {
+        "string": "Hello, World!",
+        "date_str": "2024-08-08T09:57:35.300858",
+        "datetime": time_now.isoformat(),
+        "decimal": 10.465,
+    }
+
+    admin_client.post(f"dags/{test_dag_id}/trigger", data={"conf": json.dumps(conf_dict, cls=WebEncoder)})
+
+    with create_session() as session:
+        run = session.query(DagRun).filter(DagRun.dag_id == test_dag_id).first()
+    assert run is not None
+    assert DagRunType.MANUAL in run.run_id
+    assert run.run_type == DagRunType.MANUAL
+    assert run.conf == expected_conf
+
+
 def test_trigger_dag_conf_malformed(admin_client):
     test_dag_id = "example_bash_operator"