shahar1 commented on code in PR #66156:
URL: https://github.com/apache/airflow/pull/66156#discussion_r3170259305
##########
providers/google/src/airflow/providers/google/cloud/operators/pubsub.py:
##########
@@ -871,12 +876,22 @@ def execute_complete(self, context: Context, event:
dict[str, Any]):
if event["status"] == "success":
self.log.info("Sensor pulls messages: %s", event["message"])
messages_callback = self.messages_callback or
self._default_message_callback
- _return_value = messages_callback(event["message"], context)
+ received_messages =
self._convert_to_received_messages(event["message"])
+ _return_value = messages_callback(received_messages, context)
return _return_value
self.log.info("Sensor failed: %s", event["message"])
raise AirflowException(event["message"])
+ def _convert_to_received_messages(self, messages: Any) ->
list[ReceivedMessage]:
+ try:
+ received_messages = [pubsub_v1.types.ReceivedMessage(msg) for msg
in messages]
+ return received_messages
+ except Exception as e:
+ raise PubSubMessageTransformException(
+ f"Error converting triggerer event message back to received
message format: {e}"
+ )
Review Comment:
We try to avoid usage of customized exception - could we remove
`PubSubMessageTransformException` and use a native Python exception instead?
cc: @Lee-W - we might want to avoid creation of new derivatives as part of
the pre-commit.
--
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]
