ashb commented on code in PR #24038: URL: https://github.com/apache/airflow/pull/24038#discussion_r886541057
########## 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 ValueError("messages list cannot be empty") ``` -- 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]
