ashb commented on code in PR #63801:
URL: https://github.com/apache/airflow/pull/63801#discussion_r3063648298
##########
airflow-core/src/airflow/api_fastapi/common/exceptions.py:
##########
@@ -122,4 +123,101 @@ def exception_handler(self, request: Request, exc:
DeserializationError):
)
+class ExecutionHTTPException(HTTPException):
Review Comment:
Why is this in common if it's specific to the Execution API?
##########
airflow-core/src/airflow/api_fastapi/common/exceptions.py:
##########
@@ -122,4 +123,101 @@ def exception_handler(self, request: Request, exc:
DeserializationError):
)
+class ExecutionHTTPException(HTTPException):
+ """
+ HTTPException subclass used by Execution API.
+
+ Enforces consistent error response format containing `reason` and
`message` keys.
+ """
+
+ def __init__(
+ self,
+ status_code: int,
+ *,
+ reason: str,
+ message: str,
+ extra: dict[str, Any] | None = None,
+ headers: dict[str, str] | None = None,
+ ) -> None:
+ """
+ Initialize with explicit reason/message and optional extra fields.
+
+ detail will be constructed as a dict: {"reason": reason, "message":
message, **extra}
+ """
+ detail: dict[str, Any] = {"reason": reason, "message": message}
+ if extra:
+ # Do not allow overriding reason/message through extra
accidentally.
+ for k, v in extra.items():
+ if k not in ("reason", "message"):
+ detail[k] = v
+ super().__init__(status_code=status_code, detail=detail,
headers=headers)
+
+
ERROR_HANDLERS: list[BaseErrorHandler] = [_UniqueConstraintErrorHandler(),
DagErrorHandler()]
+
+
+# --------------------
+# Global exception normalization utilities and registration
+# --------------------
+
+
+def _ensure_detail_dict(detail: dict[str, Any] | str | None) -> dict[str, Any]:
+ """Normalize detail into a dict with at least reason/message keys."""
+ if isinstance(detail, str) or detail is None:
+ return {"reason": "error", "message": detail or "An error occurred"}
+ if isinstance(detail, dict):
+ # Flatten legacy/nested payloads: {"detail": "..."} -> {"reason":
"error", "message": "..."}
+ if set(detail.keys()) == {"detail"}:
+ nested_detail = detail["detail"]
+ if isinstance(nested_detail, dict):
+ normalized = dict(nested_detail)
+ normalized.setdefault("reason", "error")
+ normalized.setdefault("message", "An error occurred")
+ return normalized
+ return {"reason": "error", "message": str(nested_detail)}
+
+ # Idempotent path: keep already structured details intact.
+ normalized = dict(detail)
+ normalized.setdefault("reason", "error")
+ normalized.setdefault("message", "An error occurred")
+ return normalized
+ return {"reason": "error", "message": str(detail)}
+
+
+def register_exception_handlers(app: FastAPI) -> None:
+ """
+ Register global and specific exception handlers on the FastAPI app.
+
+ Guarantees JSON error responses carry a `detail` object with `reason` and
`message`.
+ """
+ # Specific handlers remain in place (e.g., DB unique constraint, Dag
errors)
+ for handler in ERROR_HANDLERS:
+ app.add_exception_handler(handler.exception_cls,
handler.exception_handler) # type: ignore[arg-type]
+
+ # Normalize any HTTPException to have dict detail with reason/message
+ @app.exception_handler(HTTPException)
+ async def _http_exception_handler(request: Request, exc: HTTPException):
Review Comment:
This probably shouldn't be done in line here, but put in ERROR_HANDLERS
instead -- makes register fn single purpose and much smaller.
##########
airflow-core/src/airflow/api_fastapi/common/exceptions.py:
##########
@@ -122,4 +123,101 @@ def exception_handler(self, request: Request, exc:
DeserializationError):
)
+class ExecutionHTTPException(HTTPException):
+ """
+ HTTPException subclass used by Execution API.
+
+ Enforces consistent error response format containing `reason` and
`message` keys.
+ """
+
+ def __init__(
+ self,
+ status_code: int,
+ *,
+ reason: str,
+ message: str,
+ extra: dict[str, Any] | None = None,
+ headers: dict[str, str] | None = None,
+ ) -> None:
+ """
+ Initialize with explicit reason/message and optional extra fields.
+
+ detail will be constructed as a dict: {"reason": reason, "message":
message, **extra}
+ """
+ detail: dict[str, Any] = {"reason": reason, "message": message}
+ if extra:
+ # Do not allow overriding reason/message through extra
accidentally.
+ for k, v in extra.items():
+ if k not in ("reason", "message"):
+ detail[k] = v
Review Comment:
You could do this by having extra be `**extra`, then this block can go.
##########
airflow-core/src/airflow/api_fastapi/common/exceptions.py:
##########
@@ -122,4 +123,101 @@ def exception_handler(self, request: Request, exc:
DeserializationError):
)
+class ExecutionHTTPException(HTTPException):
+ """
+ HTTPException subclass used by Execution API.
+
+ Enforces consistent error response format containing `reason` and
`message` keys.
+ """
+
+ def __init__(
+ self,
+ status_code: int,
+ *,
+ reason: str,
+ message: str,
+ extra: dict[str, Any] | None = None,
+ headers: dict[str, str] | None = None,
+ ) -> None:
+ """
+ Initialize with explicit reason/message and optional extra fields.
+
+ detail will be constructed as a dict: {"reason": reason, "message":
message, **extra}
+ """
+ detail: dict[str, Any] = {"reason": reason, "message": message}
+ if extra:
+ # Do not allow overriding reason/message through extra
accidentally.
+ for k, v in extra.items():
+ if k not in ("reason", "message"):
+ detail[k] = v
+ super().__init__(status_code=status_code, detail=detail,
headers=headers)
+
+
ERROR_HANDLERS: list[BaseErrorHandler] = [_UniqueConstraintErrorHandler(),
DagErrorHandler()]
+
+
+# --------------------
+# Global exception normalization utilities and registration
+# --------------------
+
+
+def _ensure_detail_dict(detail: dict[str, Any] | str | None) -> dict[str, Any]:
+ """Normalize detail into a dict with at least reason/message keys."""
+ if isinstance(detail, str) or detail is None:
+ return {"reason": "error", "message": detail or "An error occurred"}
+ if isinstance(detail, dict):
+ # Flatten legacy/nested payloads: {"detail": "..."} -> {"reason":
"error", "message": "..."}
+ if set(detail.keys()) == {"detail"}:
+ nested_detail = detail["detail"]
+ if isinstance(nested_detail, dict):
+ normalized = dict(nested_detail)
+ normalized.setdefault("reason", "error")
+ normalized.setdefault("message", "An error occurred")
+ return normalized
+ return {"reason": "error", "message": str(nested_detail)}
Review Comment:
Why force it to a string.
This entire function feels like backcompat that we shouldn't need given we
are in control of all of the callers.
##########
airflow-core/src/airflow/api_fastapi/common/exceptions.py:
##########
@@ -122,4 +123,101 @@ def exception_handler(self, request: Request, exc:
DeserializationError):
)
+class ExecutionHTTPException(HTTPException):
+ """
+ HTTPException subclass used by Execution API.
+
+ Enforces consistent error response format containing `reason` and
`message` keys.
+ """
+
+ def __init__(
+ self,
+ status_code: int,
+ *,
+ reason: str,
+ message: str,
+ extra: dict[str, Any] | None = None,
+ headers: dict[str, str] | None = None,
+ ) -> None:
+ """
+ Initialize with explicit reason/message and optional extra fields.
+
+ detail will be constructed as a dict: {"reason": reason, "message":
message, **extra}
+ """
+ detail: dict[str, Any] = {"reason": reason, "message": message}
+ if extra:
+ # Do not allow overriding reason/message through extra
accidentally.
+ for k, v in extra.items():
+ if k not in ("reason", "message"):
+ detail[k] = v
+ super().__init__(status_code=status_code, detail=detail,
headers=headers)
+
+
ERROR_HANDLERS: list[BaseErrorHandler] = [_UniqueConstraintErrorHandler(),
DagErrorHandler()]
+
+
+# --------------------
+# Global exception normalization utilities and registration
+# --------------------
Review Comment:
No. No hrules like this. Worst habbit of LLM code generation.
##########
airflow-core/src/airflow/api_fastapi/common/exceptions.py:
##########
@@ -122,4 +123,101 @@ def exception_handler(self, request: Request, exc:
DeserializationError):
)
+class ExecutionHTTPException(HTTPException):
+ """
+ HTTPException subclass used by Execution API.
+
+ Enforces consistent error response format containing `reason` and
`message` keys.
+ """
+
+ def __init__(
+ self,
+ status_code: int,
+ *,
+ reason: str,
+ message: str,
+ extra: dict[str, Any] | None = None,
+ headers: dict[str, str] | None = None,
+ ) -> None:
+ """
+ Initialize with explicit reason/message and optional extra fields.
+
+ detail will be constructed as a dict: {"reason": reason, "message":
message, **extra}
+ """
+ detail: dict[str, Any] = {"reason": reason, "message": message}
+ if extra:
+ # Do not allow overriding reason/message through extra
accidentally.
+ for k, v in extra.items():
+ if k not in ("reason", "message"):
+ detail[k] = v
+ super().__init__(status_code=status_code, detail=detail,
headers=headers)
+
+
ERROR_HANDLERS: list[BaseErrorHandler] = [_UniqueConstraintErrorHandler(),
DagErrorHandler()]
+
+
+# --------------------
+# Global exception normalization utilities and registration
+# --------------------
+
+
+def _ensure_detail_dict(detail: dict[str, Any] | str | None) -> dict[str, Any]:
+ """Normalize detail into a dict with at least reason/message keys."""
+ if isinstance(detail, str) or detail is None:
+ return {"reason": "error", "message": detail or "An error occurred"}
+ if isinstance(detail, dict):
+ # Flatten legacy/nested payloads: {"detail": "..."} -> {"reason":
"error", "message": "..."}
+ if set(detail.keys()) == {"detail"}:
+ nested_detail = detail["detail"]
+ if isinstance(nested_detail, dict):
+ normalized = dict(nested_detail)
+ normalized.setdefault("reason", "error")
+ normalized.setdefault("message", "An error occurred")
+ return normalized
+ return {"reason": "error", "message": str(nested_detail)}
+
+ # Idempotent path: keep already structured details intact.
+ normalized = dict(detail)
+ normalized.setdefault("reason", "error")
+ normalized.setdefault("message", "An error occurred")
+ return normalized
+ return {"reason": "error", "message": str(detail)}
+
+
+def register_exception_handlers(app: FastAPI) -> None:
+ """
+ Register global and specific exception handlers on the FastAPI app.
+
+ Guarantees JSON error responses carry a `detail` object with `reason` and
`message`.
+ """
+ # Specific handlers remain in place (e.g., DB unique constraint, Dag
errors)
+ for handler in ERROR_HANDLERS:
+ app.add_exception_handler(handler.exception_cls,
handler.exception_handler) # type: ignore[arg-type]
+
+ # Normalize any HTTPException to have dict detail with reason/message
+ @app.exception_handler(HTTPException)
+ async def _http_exception_handler(request: Request, exc: HTTPException):
+ detail = _ensure_detail_dict(getattr(exc, "detail", None))
+ headers = getattr(exc, "headers", None)
+ return JSONResponse(status_code=exc.status_code, content={"detail":
detail}, headers=headers)
+
+ # Catch-all for unhandled Exceptions
+ @app.exception_handler(Exception)
+ async def _unhandled_exception_handler(request: Request, exc: Exception):
+ exception_id = get_random_string()
Review Comment:
What is the point of this? Is it logged anywhere to be able to correlate it
against the exception when it's not sent to the client.
##########
airflow-core/src/airflow/api_fastapi/common/exceptions.py:
##########
@@ -122,4 +123,101 @@ def exception_handler(self, request: Request, exc:
DeserializationError):
)
+class ExecutionHTTPException(HTTPException):
+ """
+ HTTPException subclass used by Execution API.
+
+ Enforces consistent error response format containing `reason` and
`message` keys.
+ """
+
+ def __init__(
+ self,
+ status_code: int,
+ *,
+ reason: str,
+ message: str,
+ extra: dict[str, Any] | None = None,
+ headers: dict[str, str] | None = None,
+ ) -> None:
+ """
+ Initialize with explicit reason/message and optional extra fields.
+
+ detail will be constructed as a dict: {"reason": reason, "message":
message, **extra}
+ """
+ detail: dict[str, Any] = {"reason": reason, "message": message}
+ if extra:
+ # Do not allow overriding reason/message through extra
accidentally.
+ for k, v in extra.items():
+ if k not in ("reason", "message"):
+ detail[k] = v
+ super().__init__(status_code=status_code, detail=detail,
headers=headers)
+
+
ERROR_HANDLERS: list[BaseErrorHandler] = [_UniqueConstraintErrorHandler(),
DagErrorHandler()]
+
+
+# --------------------
+# Global exception normalization utilities and registration
+# --------------------
+
+
+def _ensure_detail_dict(detail: dict[str, Any] | str | None) -> dict[str, Any]:
+ """Normalize detail into a dict with at least reason/message keys."""
+ if isinstance(detail, str) or detail is None:
+ return {"reason": "error", "message": detail or "An error occurred"}
+ if isinstance(detail, dict):
+ # Flatten legacy/nested payloads: {"detail": "..."} -> {"reason":
"error", "message": "..."}
+ if set(detail.keys()) == {"detail"}:
+ nested_detail = detail["detail"]
+ if isinstance(nested_detail, dict):
+ normalized = dict(nested_detail)
+ normalized.setdefault("reason", "error")
+ normalized.setdefault("message", "An error occurred")
+ return normalized
+ return {"reason": "error", "message": str(nested_detail)}
+
+ # Idempotent path: keep already structured details intact.
+ normalized = dict(detail)
Review Comment:
Why do we need a new dict?
##########
airflow-core/src/airflow/api_fastapi/common/exceptions.py:
##########
@@ -122,4 +123,101 @@ def exception_handler(self, request: Request, exc:
DeserializationError):
)
+class ExecutionHTTPException(HTTPException):
+ """
+ HTTPException subclass used by Execution API.
+
+ Enforces consistent error response format containing `reason` and
`message` keys.
+ """
+
+ def __init__(
+ self,
+ status_code: int,
+ *,
+ reason: str,
+ message: str,
+ extra: dict[str, Any] | None = None,
+ headers: dict[str, str] | None = None,
+ ) -> None:
+ """
+ Initialize with explicit reason/message and optional extra fields.
+
+ detail will be constructed as a dict: {"reason": reason, "message":
message, **extra}
+ """
+ detail: dict[str, Any] = {"reason": reason, "message": message}
+ if extra:
+ # Do not allow overriding reason/message through extra
accidentally.
+ for k, v in extra.items():
+ if k not in ("reason", "message"):
+ detail[k] = v
+ super().__init__(status_code=status_code, detail=detail,
headers=headers)
+
+
ERROR_HANDLERS: list[BaseErrorHandler] = [_UniqueConstraintErrorHandler(),
DagErrorHandler()]
+
+
+# --------------------
+# Global exception normalization utilities and registration
+# --------------------
+
+
+def _ensure_detail_dict(detail: dict[str, Any] | str | None) -> dict[str, Any]:
+ """Normalize detail into a dict with at least reason/message keys."""
+ if isinstance(detail, str) or detail is None:
+ return {"reason": "error", "message": detail or "An error occurred"}
+ if isinstance(detail, dict):
+ # Flatten legacy/nested payloads: {"detail": "..."} -> {"reason":
"error", "message": "..."}
+ if set(detail.keys()) == {"detail"}:
+ nested_detail = detail["detail"]
+ if isinstance(nested_detail, dict):
+ normalized = dict(nested_detail)
+ normalized.setdefault("reason", "error")
+ normalized.setdefault("message", "An error occurred")
+ return normalized
+ return {"reason": "error", "message": str(nested_detail)}
+
+ # Idempotent path: keep already structured details intact.
+ normalized = dict(detail)
+ normalized.setdefault("reason", "error")
+ normalized.setdefault("message", "An error occurred")
+ return normalized
+ return {"reason": "error", "message": str(detail)}
+
+
+def register_exception_handlers(app: FastAPI) -> None:
+ """
+ Register global and specific exception handlers on the FastAPI app.
+
+ Guarantees JSON error responses carry a `detail` object with `reason` and
`message`.
+ """
+ # Specific handlers remain in place (e.g., DB unique constraint, Dag
errors)
+ for handler in ERROR_HANDLERS:
+ app.add_exception_handler(handler.exception_cls,
handler.exception_handler) # type: ignore[arg-type]
+
+ # Normalize any HTTPException to have dict detail with reason/message
+ @app.exception_handler(HTTPException)
+ async def _http_exception_handler(request: Request, exc: HTTPException):
+ detail = _ensure_detail_dict(getattr(exc, "detail", None))
+ headers = getattr(exc, "headers", None)
+ return JSONResponse(status_code=exc.status_code, content={"detail":
detail}, headers=headers)
+
+ # Catch-all for unhandled Exceptions
+ @app.exception_handler(Exception)
+ async def _unhandled_exception_handler(request: Request, exc: Exception):
+ exception_id = get_random_string()
+ log.exception("Unhandled error id %s while handling request %s",
exception_id, request.url.path)
+
+ if conf.get("api", "expose_stacktrace") == "True":
+ message = f"Unhandled error id {exception_id}: {exc}"
Review Comment:
Give this is JSON, we shouldn't put multiple things into a single string
should we -- we've got structure, lets use it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
