[v3-2-test] UI: Return 400 instead of 500 from structure_data on malformed asset_expression (#67489)

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -788,6 +788,12 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/StructureDataResponse'
+        '400':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Bad Request
         '404':
           content:
             application/json:

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/structure.py

@@ -42,7 +42,12 @@
 
 @structure_router.get(
     "/structure_data",
-    responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    responses=create_openapi_http_exception_doc(
+        [
+            status.HTTP_400_BAD_REQUEST,
+            status.HTTP_404_NOT_FOUND,
+        ]
+    ),
     dependencies=[
         Depends(requires_access_dag("GET")),
         Depends(requires_access_dag("GET", DagAccessEntity.DEPENDENCIES)),
@@ -161,9 +166,15 @@ def structure_data(
                 )
 
         if (asset_expression := serialized_dag.dag_model.asset_expression) and entry_node_ref:
-            upstream_asset_nodes, upstream_asset_edges = get_upstream_assets(
-                asset_expression, entry_node_ref["id"]
-            )
+            try:
+                upstream_asset_nodes, upstream_asset_edges = get_upstream_assets(
+                    asset_expression, entry_node_ref["id"]
+                )
+            except TypeError as e:
+                raise HTTPException(
+                    status.HTTP_400_BAD_REQUEST,
+                    f"Malformed asset_expression in Dag {dag_id!r} version {version_number}: {e}",
+                ) from e
             data["nodes"] += upstream_asset_nodes
             data["edges"] += upstream_asset_edges
 

airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts

@@ -4256,6 +4256,7 @@ export class StructureService {
                 version_number: data.versionNumber
             },
             errors: {
+                400: 'Bad Request',
                 404: 'Not Found',
                 422: 'Validation Error'
             }

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -7254,6 +7254,10 @@ export type $OpenApiTs = {
                  * Successful Response
                  */
                 200: StructureDataResponse;
+                /**
+                 * Bad Request
+                 */
+                400: HTTPExceptionResponse;
                 /**
                  * Not Found
                  */

airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_structure.py

@@ -709,6 +709,31 @@ def test_should_return_404(self, test_client):
         assert response.status_code == 404
         assert response.json()["detail"] == "Dag with id not_existing was not found"
 
+    @pytest.mark.usefixtures("make_dags")
+    def test_should_return_400_on_malformed_asset_expression(self, test_client):
+        """A TypeError from get_upstream_assets surfaces as a 400 with a clear message.
+
+        The asset_expression ultimately comes from user-authored Dag code (via the Task SDK),
+        so a malformed expression is bad input that ended up persisted -- not a server fault.
+        Without the try/except wrap, the TypeError propagates uncaught and FastAPI returns a
+        generic ``{"detail": "Internal Server Error"}`` 500 body with no context about which
+        Dag triggered it. With the wrap, the response identifies the Dag and version, which
+        is what a caller needs to fix the upstream Dag definition.
+        """
+        with mock.patch(
+            "airflow.api_fastapi.core_api.routes.ui.structure.get_upstream_assets",
+            side_effect=TypeError("Unsupported type: dict_keys(['weird-op'])"),
+        ):
+            response = test_client.get(
+                "/structure/structure_data",
+                params={"dag_id": DAG_ID, "external_dependencies": True},
+            )
+        assert response.status_code == 400
+        detail = response.json()["detail"]
+        assert "Malformed asset_expression" in detail
+        assert DAG_ID in detail
+        assert "Unsupported type" in detail
+
     def test_should_return_404_when_dag_version_not_found(self, test_client):
         response = test_client.get(
             "/structure/structure_data", params={"dag_id": DAG_ID, "version_number": 999}