jason810496 commented on code in PR #49164:
URL: https://github.com/apache/airflow/pull/49164#discussion_r2264876400
##########
airflow-core/tests/unit/api_fastapi/common/test_exceptions.py:
##########
@@ -282,3 +294,102 @@ def test_handle_multiple_columns_unique_constraint_error(
assert response_detail == expected_detail
assert "INSERT INTO dag_run" in actual_statement
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+
+class TestDAGErrorHandler:
+ dag_error_handler = DAGErrorHandler()
+
+ def test_handle_deserialization_error_with_value_error(self):
+ error_message = "Missing DAG ID in serialized DAG"
+ deserialization_error = DeserializationError("test_dag_id")
+
+ value_error = ValueError(error_message)
+ deserialization_error.__cause__ = value_error
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{deserialization_error}",
+ )
+
+ with pytest.raises(HTTPException) as exeinfo_response_error:
+ self.dag_error_handler.exception_handler(None,
deserialization_error)
+
+ assert exeinfo_response_error.value.status_code ==
expected_exception.status_code
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+ def test_handle_deserialization_error_with_runtime_error(self):
+ error_message = "Error during DAG serialization process"
+ deserialization_error = DeserializationError("test_dag_id")
+
+ runtime_error = RuntimeError(error_message)
+ deserialization_error.__cause__ = runtime_error
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{deserialization_error}",
+ )
+
+ with pytest.raises(HTTPException) as exeinfo_response_error:
+ self.dag_error_handler.exception_handler(None,
deserialization_error)
+
+ assert exeinfo_response_error.value.status_code ==
expected_exception.status_code
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+ def test_handle_deserialization_error_with_key_error(self):
+ key = "required_field"
+ deserialization_error = DeserializationError("test_dag_id")
+
+ key_error = KeyError(key)
+ deserialization_error.__cause__ = key_error
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{deserialization_error}",
+ )
+
+ with pytest.raises(HTTPException) as exeinfo_response_error:
+ self.dag_error_handler.exception_handler(None,
deserialization_error)
+
+ assert exeinfo_response_error.value.status_code ==
expected_exception.status_code
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+ def test_handle_real_dag_deserialization_error(self):
+ """Test handling a real DAG deserialization error with actual
serialized DAG."""
+
+ # Create a test DAG
+ dag_id = "test_dag_for_error"
+ test_dag = DAG(dag_id=dag_id, start_date=datetime(2025, 4, 15),
schedule="@once")
+ EmptyOperator(task_id="test_task", dag=test_dag)
+ test_dag.sync_to_db()
+ SerializedDagModel.write_dag(test_dag, bundle_name=dag_id)
+
+ with create_session() as session:
+ dag_model =
session.scalar(select(SerializedDagModel).where(SerializedDagModel.dag_id ==
dag_id))
+ if not dag_model:
+ pytest.fail("Failed to find serialized DAG in database")
+ data = dag_model.data
+ del data["dag"]["dag_id"]
+ session.execute(
+ update(SerializedDagModel).where(SerializedDagModel.dag_id ==
dag_id).values(_data=data)
+ )
+ session.commit()
+
+ dag_bag = DagBag(read_dags_from_db=True)
+ with pytest.raises(DeserializationError) as exc_info:
+ dag_bag.get_dag(dag_id)
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{exc_info.value}",
Review Comment:
```suggestion
detail=f"An error occurred while trying to deserialize Dag:
{exc_info.value}",
```
##########
airflow-core/tests/unit/api_fastapi/common/test_exceptions.py:
##########
@@ -282,3 +294,102 @@ def test_handle_multiple_columns_unique_constraint_error(
assert response_detail == expected_detail
assert "INSERT INTO dag_run" in actual_statement
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+
+class TestDAGErrorHandler:
+ dag_error_handler = DAGErrorHandler()
+
+ def test_handle_deserialization_error_with_value_error(self):
+ error_message = "Missing DAG ID in serialized DAG"
+ deserialization_error = DeserializationError("test_dag_id")
+
+ value_error = ValueError(error_message)
+ deserialization_error.__cause__ = value_error
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{deserialization_error}",
+ )
+
+ with pytest.raises(HTTPException) as exeinfo_response_error:
+ self.dag_error_handler.exception_handler(None,
deserialization_error)
+
+ assert exeinfo_response_error.value.status_code ==
expected_exception.status_code
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+ def test_handle_deserialization_error_with_runtime_error(self):
+ error_message = "Error during DAG serialization process"
+ deserialization_error = DeserializationError("test_dag_id")
+
+ runtime_error = RuntimeError(error_message)
+ deserialization_error.__cause__ = runtime_error
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{deserialization_error}",
+ )
+
+ with pytest.raises(HTTPException) as exeinfo_response_error:
+ self.dag_error_handler.exception_handler(None,
deserialization_error)
+
+ assert exeinfo_response_error.value.status_code ==
expected_exception.status_code
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+ def test_handle_deserialization_error_with_key_error(self):
+ key = "required_field"
+ deserialization_error = DeserializationError("test_dag_id")
+
+ key_error = KeyError(key)
+ deserialization_error.__cause__ = key_error
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{deserialization_error}",
+ )
+
+ with pytest.raises(HTTPException) as exeinfo_response_error:
+ self.dag_error_handler.exception_handler(None,
deserialization_error)
+
+ assert exeinfo_response_error.value.status_code ==
expected_exception.status_code
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+ def test_handle_real_dag_deserialization_error(self):
+ """Test handling a real DAG deserialization error with actual
serialized DAG."""
+
+ # Create a test DAG
+ dag_id = "test_dag_for_error"
+ test_dag = DAG(dag_id=dag_id, start_date=datetime(2025, 4, 15),
schedule="@once")
+ EmptyOperator(task_id="test_task", dag=test_dag)
+ test_dag.sync_to_db()
+ SerializedDagModel.write_dag(test_dag, bundle_name=dag_id)
+
+ with create_session() as session:
+ dag_model =
session.scalar(select(SerializedDagModel).where(SerializedDagModel.dag_id ==
dag_id))
+ if not dag_model:
+ pytest.fail("Failed to find serialized DAG in database")
Review Comment:
```suggestion
pytest.fail("Failed to find serialized Dag in database")
```
##########
airflow-core/src/airflow/api_fastapi/common/exceptions.py:
##########
@@ -103,6 +105,18 @@ def _is_dialect_matched(self, exc: IntegrityError) -> bool:
return False
-DatabaseErrorHandlers = [
- _UniqueConstraintErrorHandler(),
-]
+class DAGErrorHandler(BaseErrorHandler[DeserializationError]):
+ """Handler for DAG-related errors."""
+
+ def __init__(self):
+ super().__init__(DeserializationError)
+
+ def exception_handler(self, request: Request, exc: DeserializationError):
+ """Handle DAG deserialization exceptions."""
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail=f"An error occurred while trying to deserialize DAG: {exc}",
+ )
+
+
+ERROR_HANDLERS = [_UniqueConstraintErrorHandler(), DAGErrorHandler()]
Review Comment:
```suggestion
class DagErrorHandler(BaseErrorHandler[DeserializationError]):
"""Handler for Dag-related errors."""
def __init__(self):
super().__init__(DeserializationError)
def exception_handler(self, request: Request, exc: DeserializationError):
"""Handle Dag deserialization exceptions."""
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"An error occurred while trying to deserialize Dag:
{exc}",
)
ERROR_HANDLERS = [_UniqueConstraintErrorHandler(), DagErrorHandler()]
```
##########
airflow-core/src/airflow/exceptions.py:
##########
@@ -517,6 +517,23 @@ class UnknownExecutorException(ValueError):
"""Raised when an attempt is made to load an executor which is not
configured."""
+class DeserializationError(Exception):
+ """
+ Raised when a DAG cannot be deserialized.
+
+ This exception should be raised using exception chaining:
+ `raise DeserializationError(dag_id) from original_exception`
+ """
+
+ def __init__(self, dag_id: str = "unknown"):
+ self.dag_id = dag_id
+ if dag_id == "unknown":
+ message = "Missing DAG ID in serialized DAG"
+ else:
+ message = f"An unexpected error occurred while trying to
deserialize DAG '{dag_id}'"
+ super().__init__(message)
+
Review Comment:
```suggestion
class DeserializationError(Exception):
"""
Raised when a Dag cannot be deserialized.
This exception should be raised using exception chaining:
`raise DeserializationError(dag_id) from original_exception`
"""
def __init__(self, dag_id: str = "unknown"):
self.dag_id = dag_id
if dag_id == "unknown":
message = "Missing Dag ID in serialized Dag"
else:
message = f"An unexpected error occurred while trying to
deserialize Dag '{dag_id}'"
super().__init__(message)
```
##########
airflow-core/tests/unit/api_fastapi/common/test_exceptions.py:
##########
@@ -282,3 +294,102 @@ def test_handle_multiple_columns_unique_constraint_error(
assert response_detail == expected_detail
assert "INSERT INTO dag_run" in actual_statement
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+
+class TestDAGErrorHandler:
+ dag_error_handler = DAGErrorHandler()
+
+ def test_handle_deserialization_error_with_value_error(self):
+ error_message = "Missing DAG ID in serialized DAG"
+ deserialization_error = DeserializationError("test_dag_id")
+
+ value_error = ValueError(error_message)
+ deserialization_error.__cause__ = value_error
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{deserialization_error}",
Review Comment:
```suggestion
class TestDagErrorHandler:
dag_error_handler = DagErrorHandler()
def test_handle_deserialization_error_with_value_error(self):
error_message = "Missing Dag ID in serialized Dag"
deserialization_error = DeserializationError("test_dag_id")
value_error = ValueError(error_message)
deserialization_error.__cause__ = value_error
expected_exception = HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"An error occurred while trying to deserialize Dag:
{deserialization_error}",
```
##########
airflow-core/tests/unit/api_fastapi/common/test_exceptions.py:
##########
@@ -282,3 +294,102 @@ def test_handle_multiple_columns_unique_constraint_error(
assert response_detail == expected_detail
assert "INSERT INTO dag_run" in actual_statement
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+
+class TestDAGErrorHandler:
+ dag_error_handler = DAGErrorHandler()
+
+ def test_handle_deserialization_error_with_value_error(self):
+ error_message = "Missing DAG ID in serialized DAG"
+ deserialization_error = DeserializationError("test_dag_id")
+
+ value_error = ValueError(error_message)
+ deserialization_error.__cause__ = value_error
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{deserialization_error}",
+ )
+
+ with pytest.raises(HTTPException) as exeinfo_response_error:
+ self.dag_error_handler.exception_handler(None,
deserialization_error)
+
+ assert exeinfo_response_error.value.status_code ==
expected_exception.status_code
+ assert exeinfo_response_error.value.detail == expected_exception.detail
+
+ def test_handle_deserialization_error_with_runtime_error(self):
+ error_message = "Error during DAG serialization process"
+ deserialization_error = DeserializationError("test_dag_id")
+
+ runtime_error = RuntimeError(error_message)
+ deserialization_error.__cause__ = runtime_error
+
+ expected_exception = HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"An error occurred while trying to deserialize DAG:
{deserialization_error}",
Review Comment:
```suggestion
error_message = "Error during Dag serialization process"
deserialization_error = DeserializationError("test_dag_id")
runtime_error = RuntimeError(error_message)
deserialization_error.__cause__ = runtime_error
expected_exception = HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"An error occurred while trying to deserialize Dag:
{deserialization_error}",
```
--
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]
