You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/06/01 08:44:22 UTC
[GitHub] [airflow] ashb commented on a diff in pull request #24038: Implement Azure Service Bus Queue Operator's
ashb commented on code in PR #24038:
URL: https://github.com/apache/airflow/pull/24038#discussion_r886540605
##########
airflow/providers/microsoft/azure/hooks/asb_message.py:
##########
@@ -0,0 +1,122 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import List, Optional, Union
+
+from azure.servicebus import ServiceBusClient, ServiceBusMessage, ServiceBusSender
+from azure.servicebus.exceptions import MessageSizeExceededError
+
+from airflow.exceptions import AirflowBadRequest, AirflowException
+from airflow.providers.microsoft.azure.hooks.base_asb import BaseAzureServiceBusHook
+
+
+class ServiceBusMessageHook(BaseAzureServiceBusHook):
+ """
+ Interacts with ServiceBusClient and acts as a high level interface
+ for getting ServiceBusSender and ServiceBusReceiver.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+
+ def get_conn(self) -> ServiceBusClient:
+ """Create and returns ServiceBusClient by using the connection string in connection details"""
+ conn = self.get_connection(self.conn_id)
+ extras = conn.extra_dejson
+
+ self.connection_string = str(
+ extras.get('connection_string') or extras.get('extra__azure_service_bus__connection_string')
+ )
+
+ return ServiceBusClient.from_connection_string(conn_str=self.connection_string, logging_enable=True)
+
+ def send_message(
+ self, queue_name: str, messages: Union[str, List[str]], batch_message_flag: bool = False
+ ):
+ """
+ By using ServiceBusClient Send message(s) to a Service Bus Queue. By using
+ batch_message_flag it enables and send message as batch message
+
+ :param queue_name: The name of the queue or a QueueProperties with name.
+ :param messages: Message which needs to be sent to the queue. It can be string or list of string.
+ :param batch_message_flag: bool flag, can be set to True if message needs to be sent as batch message.
+ """
+ if queue_name is None:
+ raise AirflowBadRequest("Queue name cannot be None.")
Review Comment:
```suggestion
raise TypeError("Queue name cannot be None.")
```
##########
airflow/providers/microsoft/azure/hooks/asb_message.py:
##########
@@ -0,0 +1,122 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import List, Optional, Union
+
+from azure.servicebus import ServiceBusClient, ServiceBusMessage, ServiceBusSender
+from azure.servicebus.exceptions import MessageSizeExceededError
+
+from airflow.exceptions import AirflowBadRequest, AirflowException
+from airflow.providers.microsoft.azure.hooks.base_asb import BaseAzureServiceBusHook
+
+
+class ServiceBusMessageHook(BaseAzureServiceBusHook):
+ """
+ Interacts with ServiceBusClient and acts as a high level interface
+ for getting ServiceBusSender and ServiceBusReceiver.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+
+ def get_conn(self) -> ServiceBusClient:
+ """Create and returns ServiceBusClient by using the connection string in connection details"""
+ conn = self.get_connection(self.conn_id)
+ extras = conn.extra_dejson
+
+ self.connection_string = str(
+ extras.get('connection_string') or extras.get('extra__azure_service_bus__connection_string')
+ )
+
+ return ServiceBusClient.from_connection_string(conn_str=self.connection_string, logging_enable=True)
+
+ def send_message(
+ self, queue_name: str, messages: Union[str, List[str]], batch_message_flag: bool = False
+ ):
+ """
+ By using ServiceBusClient Send message(s) to a Service Bus Queue. By using
+ batch_message_flag it enables and send message as batch message
+
+ :param queue_name: The name of the queue or a QueueProperties with name.
+ :param messages: Message which needs to be sent to the queue. It can be string or list of string.
+ :param batch_message_flag: bool flag, can be set to True if message needs to be sent as batch message.
+ """
+ if queue_name is None:
+ raise AirflowBadRequest("Queue name cannot be None.")
+ if not messages:
+ raise AirflowException("Message is empty.")
Review Comment:
```suggestion
raise Value("messages list cannot be empty")
```
##########
airflow/providers/microsoft/azure/hooks/asb_message.py:
##########
@@ -0,0 +1,122 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import List, Optional, Union
+
+from azure.servicebus import ServiceBusClient, ServiceBusMessage, ServiceBusSender
+from azure.servicebus.exceptions import MessageSizeExceededError
+
+from airflow.exceptions import AirflowBadRequest, AirflowException
+from airflow.providers.microsoft.azure.hooks.base_asb import BaseAzureServiceBusHook
+
+
+class ServiceBusMessageHook(BaseAzureServiceBusHook):
+ """
+ Interacts with ServiceBusClient and acts as a high level interface
+ for getting ServiceBusSender and ServiceBusReceiver.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+
+ def get_conn(self) -> ServiceBusClient:
+ """Create and returns ServiceBusClient by using the connection string in connection details"""
+ conn = self.get_connection(self.conn_id)
+ extras = conn.extra_dejson
+
+ self.connection_string = str(
+ extras.get('connection_string') or extras.get('extra__azure_service_bus__connection_string')
+ )
+
+ return ServiceBusClient.from_connection_string(conn_str=self.connection_string, logging_enable=True)
+
+ def send_message(
+ self, queue_name: str, messages: Union[str, List[str]], batch_message_flag: bool = False
+ ):
+ """
+ By using ServiceBusClient Send message(s) to a Service Bus Queue. By using
+ batch_message_flag it enables and send message as batch message
+
+ :param queue_name: The name of the queue or a QueueProperties with name.
+ :param messages: Message which needs to be sent to the queue. It can be string or list of string.
+ :param batch_message_flag: bool flag, can be set to True if message needs to be sent as batch message.
+ """
+ if queue_name is None:
+ raise AirflowBadRequest("Queue name cannot be None.")
+ if not messages:
+ raise AirflowException("Message is empty.")
+ service_bus_client = self.get_conn()
+ try:
+ with service_bus_client:
+ sender = service_bus_client.get_queue_sender(queue_name=queue_name)
+ with sender:
+ if isinstance(messages, str):
+ if not batch_message_flag:
+ msg = ServiceBusMessage(messages)
+ sender.send_messages(msg)
+ else:
+ self.send_batch_message(sender, [messages])
+ else:
+ if not batch_message_flag:
+ self.send_list_messages(sender, messages)
+ else:
+ self.send_batch_message(sender, messages)
+ except Exception as e:
+ raise AirflowException(e)
Review Comment:
There is no need for this to be an AirflowException -- so lets git rid of the try/except block and just let the exception be thrown.
##########
airflow/providers/microsoft/azure/hooks/asb_message.py:
##########
@@ -0,0 +1,122 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import List, Optional, Union
+
+from azure.servicebus import ServiceBusClient, ServiceBusMessage, ServiceBusSender
+from azure.servicebus.exceptions import MessageSizeExceededError
+
+from airflow.exceptions import AirflowBadRequest, AirflowException
+from airflow.providers.microsoft.azure.hooks.base_asb import BaseAzureServiceBusHook
+
+
+class ServiceBusMessageHook(BaseAzureServiceBusHook):
+ """
+ Interacts with ServiceBusClient and acts as a high level interface
+ for getting ServiceBusSender and ServiceBusReceiver.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+
+ def get_conn(self) -> ServiceBusClient:
+ """Create and returns ServiceBusClient by using the connection string in connection details"""
+ conn = self.get_connection(self.conn_id)
+ extras = conn.extra_dejson
+
+ self.connection_string = str(
+ extras.get('connection_string') or extras.get('extra__azure_service_bus__connection_string')
+ )
+
+ return ServiceBusClient.from_connection_string(conn_str=self.connection_string, logging_enable=True)
+
+ def send_message(
+ self, queue_name: str, messages: Union[str, List[str]], batch_message_flag: bool = False
+ ):
+ """
+ By using ServiceBusClient Send message(s) to a Service Bus Queue. By using
+ batch_message_flag it enables and send message as batch message
+
+ :param queue_name: The name of the queue or a QueueProperties with name.
+ :param messages: Message which needs to be sent to the queue. It can be string or list of string.
+ :param batch_message_flag: bool flag, can be set to True if message needs to be sent as batch message.
+ """
+ if queue_name is None:
+ raise AirflowBadRequest("Queue name cannot be None.")
+ if not messages:
+ raise AirflowException("Message is empty.")
+ service_bus_client = self.get_conn()
+ try:
+ with service_bus_client:
+ sender = service_bus_client.get_queue_sender(queue_name=queue_name)
+ with sender:
Review Comment:
Can be combined in to a single `with` using multiple values.
```suggestion
with self.get_conn() as service_bus_client, service_bus_client.get_queue_sender(
queue_name=queue_name
) as sender:
```
(this includes removing the `try`/`except` - see other comment)
As a general rule, I try to keep code as un-indented as possible: the more indentation there is the more complex it is.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org