Skip to content

[ServiceBus] expand kwargs in public API #22353

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
swathipil merged 11 commits into from
Jan 8, 2022
Merged

[ServiceBus] expand kwargs in public API #22353

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
21efe9d
inital expand kwargs w/o mgmt models
swathipil Jan 6, 2022
5be09e8
fix mypy/pylint
swathipil Jan 6, 2022
03a93bb
mypy
swathipil Jan 6, 2022
9171e49
adam + anna comments
swathipil Jan 7, 2022
4983f60
Merge branch 'main' into swathipil/sb/expand-kwargs
swathipil Jan 7, 2022
ef19350
lint
swathipil Jan 7, 2022
c9e0c95
lint bad whitespace
swathipil Jan 7, 2022
7b40dbc
Merge branch 'main' into swathipil/sb/expand-kwargs
swathipil Jan 7, 2022
dc554cc
remove retry
swathipil Jan 7, 2022
6bc5129
adams comments
swathipil Jan 7, 2022
b1a646e
lint
swathipil Jan 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 16 additions & 7 deletions sdk/servicebus/azure-servicebus/azure/servicebus/_common/_connection_string_parser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

from typing import Optional
from ..management._models import DictMixin
from .._base_handler import _parse_conn_str

Expand All @@ -12,13 +13,21 @@ class ServiceBusConnectionStringProperties(DictMixin):
Properties of a connection string.
"""

def __init__(self, **kwargs):
self._fully_qualified_namespace = kwargs.pop("fully_qualified_namespace", None)
self._endpoint = kwargs.pop("endpoint", None)
self._entity_path = kwargs.pop("entity_path", None)
self._shared_access_signature = kwargs.pop("shared_access_signature", None)
self._shared_access_key_name = kwargs.pop("shared_access_key_name", None)
self._shared_access_key = kwargs.pop("shared_access_key", None)
def __init__(
self,
fully_qualified_namespace: Optional[str] = None,
endpoint: Optional[str] = None,
entity_path: Optional[str] = None,
shared_access_signature: Optional[str] = None,
shared_access_key_name: Optional[str] = None,
shared_access_key: Optional[str] = None,
Comment thread
swathipil marked this conversation as resolved.
Outdated
) -> None:
self._fully_qualified_namespace = fully_qualified_namespace
self._endpoint = endpoint
self._entity_path = entity_path
self._shared_access_signature = shared_access_signature
self._shared_access_key_name = shared_access_key_name
self._shared_access_key = shared_access_key

@property
def fully_qualified_namespace(self):
Expand Down
46 changes: 30 additions & 16 deletions sdk/servicebus/azure-servicebus/azure/servicebus/_common/message.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,8 +92,24 @@ class ServiceBusMessage(

"""

def __init__(self, body, **kwargs):
# type: (Optional[Union[str, bytes]], Any) -> None
def __init__(
self,
body: Optional[Union[str, bytes]],
*,
application_properties: Optional[dict] = None,
Comment thread
swathipil marked this conversation as resolved.
Outdated
session_id: Optional[str] = None,
message_id: Optional[str] = None,
scheduled_enqueue_time_utc: Optional[datetime.datetime] = None,
time_to_live: Optional[datetime.timedelta] = None,
content_type: Optional[str] = None,
correlation_id: Optional[str] = None,
subject: Optional[str] = None,
partition_key: Optional[str] = None,
to: Optional[str] = None,
reply_to: Optional[str] = None,
reply_to_session_id: Optional[str] = None,
**kwargs: Any
) -> None:
# Although we might normally thread through **kwargs this causes
# problems as MessageProperties won't absorb spurious args.
self._encoding = kwargs.pop("encoding", "UTF-8")
Expand All @@ -108,20 +124,18 @@ def __init__(self, body, **kwargs):
self._raw_amqp_message = AmqpAnnotatedMessage(message=self.message)
else:
self._build_message(body)
self.application_properties = kwargs.pop("application_properties", None)
self.session_id = kwargs.pop("session_id", None)
self.message_id = kwargs.pop("message_id", None)
self.content_type = kwargs.pop("content_type", None)
self.correlation_id = kwargs.pop("correlation_id", None)
self.to = kwargs.pop("to", None)
self.reply_to = kwargs.pop("reply_to", None)
self.reply_to_session_id = kwargs.pop("reply_to_session_id", None)
self.subject = kwargs.pop("subject", None)
self.scheduled_enqueue_time_utc = kwargs.pop(
"scheduled_enqueue_time_utc", None
)
self.time_to_live = kwargs.pop("time_to_live", None)
self.partition_key = kwargs.pop("partition_key", None)
self.application_properties = application_properties
self.session_id = session_id
self.message_id = message_id
self.content_type = content_type
self.correlation_id = correlation_id
self.to = to
self.reply_to = reply_to
self.reply_to_session_id = reply_to_session_id
self.subject = subject
self.scheduled_enqueue_time_utc = scheduled_enqueue_time_utc
self.time_to_live = time_to_live
self.partition_key = partition_key

def __str__(self):
# type: () -> str
Expand Down
134 changes: 116 additions & 18 deletions sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,12 @@
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from typing import Any, Union, TYPE_CHECKING
from typing import Any, Dict, Union, Optional, TYPE_CHECKING
import logging
from weakref import WeakSet

import uamqp
from uamqp.constants import TransportType
Comment thread
swathipil marked this conversation as resolved.
Outdated

from ._base_handler import (
_parse_conn_str,
Expand All @@ -15,16 +16,27 @@
)
from ._servicebus_sender import ServiceBusSender
from ._servicebus_receiver import ServiceBusReceiver
from ._common.auto_lock_renewer import AutoLockRenewer
from ._common._configuration import Configuration
from ._common.utils import (
create_authentication,
generate_dead_letter_entity_name,
strip_protocol_from_uri,
)
from ._common.constants import ServiceBusSubQueue
from ._common.constants import (
ServiceBusSubQueue,
ServiceBusReceiveMode,
ServiceBusSessionFilter
)
from ._retry import RetryMode

if TYPE_CHECKING:
from azure.core.credentials import TokenCredential, AzureSasCredential, AzureNamedKeyCredential
from azure.core.credentials import (
TokenCredential,
AzureSasCredential,
AzureNamedKeyCredential,
)


_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -74,15 +86,40 @@ class ServiceBusClient(object):

"""

def __init__(self, fully_qualified_namespace, credential, **kwargs):
# type: (str, Union[TokenCredential, AzureSasCredential, AzureNamedKeyCredential], Any) -> None
def __init__(
self,
fully_qualified_namespace: str,
credential: Union[
"TokenCredential", "AzureSasCredential", "AzureNamedKeyCredential"
],
*,
http_proxy: Optional[Dict[str, Any]] = None,
user_agent: Optional[str] = None,
logging_enable: Optional[bool] = False,
transport_type: Optional[TransportType] = TransportType.Amqp,
Comment thread
swathipil marked this conversation as resolved.
Outdated
Comment thread
swathipil marked this conversation as resolved.
Outdated
retry_total: int = 3,
retry_backoff_factor: float = 0.8,
retry_backoff_max: int = 120,
retry_mode: RetryMode = RetryMode.EXPONENTIAL,
**kwargs: Any
) -> None:
# If the user provided http:// or sb://, let's be polite and strip that.
self.fully_qualified_namespace = strip_protocol_from_uri(
fully_qualified_namespace.strip()
)

self._credential = credential
self._config = Configuration(**kwargs)
self._config = Configuration(
http_proxy=http_proxy,
user_agent=user_agent,
logging_enable=logging_enable,
transport_type=transport_type,
retry_total=retry_total,
retry_backoff_factor=retry_backoff_factor,
retry_backoff_max=retry_backoff_max,
retry_mode=retry_mode,
**kwargs
)
self._connection = None
# Optional entity name, can be the name of Queue or Topic. Intentionally not advertised, typically be needed.
self._entity_name = kwargs.get("entity_name")
Expand Down Expand Up @@ -133,8 +170,20 @@ def close(self):
self._connection.destroy()

@classmethod
def from_connection_string(cls, conn_str, **kwargs):
# type: (str, Any) -> ServiceBusClient
def from_connection_string(
cls,
conn_str: str,
*,
http_proxy: Optional[Dict[str, Any]] = None,
user_agent: Optional[str] = None,
logging_enable: bool = False,
transport_type: TransportType = TransportType.Amqp,
Comment thread
swathipil marked this conversation as resolved.
Outdated
retry_total: int = 3,
retry_backoff_factor: float = 0.8,
retry_backoff_max: int = 120,
retry_mode: RetryMode = RetryMode.EXPONENTIAL,
**kwargs: Any
) -> "ServiceBusClient":
"""
Create a ServiceBusClient from a connection string.

Expand Down Expand Up @@ -179,6 +228,14 @@ def from_connection_string(cls, conn_str, **kwargs):
fully_qualified_namespace=host,
entity_name=entity_in_conn_str or kwargs.pop("entity_name", None),
credential=credential, # type: ignore
http_proxy=http_proxy,
user_agent=user_agent,
logging_enable=logging_enable,
transport_type=transport_type,
retry_total=retry_total,
retry_backoff_factor=retry_backoff_factor,
retry_backoff_max=retry_backoff_max,
retry_mode=retry_mode,
**kwargs
)

Expand Down Expand Up @@ -225,15 +282,27 @@ def get_queue_sender(self, queue_name, **kwargs):
self._handlers.add(handler)
return handler

def get_queue_receiver(self, queue_name, **kwargs):
# type: (str, Any) -> ServiceBusReceiver
def get_queue_receiver(
self,
queue_name: str,
*,
session_id: Optional[Union[str, ServiceBusSessionFilter]] = None,

@yunhaoling yunhaoling Jan 6, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ServiceBusSessionFilter is not part of the public API surface

what mypy error are you getting with this: session_id: Union[str, NEXT_AVAILABLE_SESSION] = None? is there a link to the pipeline log?

@swathipil swathipil Jan 6, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://dev.azure.com/azure-sdk/public/_build/results?buildId=1277503&view=logs&j=b70e5e73-bbb6-5567-0939-8415943fadb9&t=e3f8f308-6e5f-58aa-3a2a-fe425958de32&l=70

@swathipil swathipil Jan 7, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

removing for now. BUT NEED TO REVISIT!!

sub_queue: Optional[Union[ServiceBusSubQueue, str]] = None,
receive_mode: Union[
ServiceBusReceiveMode, str
] = ServiceBusReceiveMode.PEEK_LOCK,
max_wait_time: Optional[float] = None,
auto_lock_renewer: Optional[AutoLockRenewer] = None,
prefetch_count: int = 0,
**kwargs: Any
) -> ServiceBusReceiver:
"""Get ServiceBusReceiver for the specific queue.

:param str queue_name: The path of specific Service Bus Queue the client connects to.
:keyword session_id: A specific session from which to receive. This must be specified for a
sessionful queue, otherwise it must be None. In order to receive messages from the next available
session, set this to ~azure.servicebus.NEXT_AVAILABLE_SESSION.
:paramtype session_id: Union[str, ~azure.servicebus.NEXT_AVAILABLE_SESSION]
:paramtype session_id: Optional[Union[str, ~azure.servicebus.NEXT_AVAILABLE_SESSION]]
Comment thread
swathipil marked this conversation as resolved.
Outdated
:keyword Optional[Union[ServiceBusSubQueue, str]] sub_queue: If specified, the subqueue this receiver will
connect to.
This includes the DEAD_LETTER and TRANSFER_DEAD_LETTER queues, holds messages that can't be delivered to any
Expand Down Expand Up @@ -278,8 +347,7 @@ def get_queue_receiver(self, queue_name, **kwargs):
"the connection string used to construct the ServiceBusClient."
)

sub_queue = kwargs.get("sub_queue", None)
if sub_queue and kwargs.get("session_id"):
if sub_queue and session_id:
raise ValueError(
"session_id and sub_queue can not be specified simultaneously. "
"To connect to the sub queue of a sessionful queue, "
Expand Down Expand Up @@ -312,6 +380,12 @@ def get_queue_receiver(self, queue_name, **kwargs):
retry_total=self._config.retry_total,
retry_backoff_factor=self._config.retry_backoff_factor,
retry_backoff_max=self._config.retry_backoff_max,
session_id=session_id,
sub_queue=sub_queue,
receive_mode=receive_mode,
max_wait_time=max_wait_time,
auto_lock_renewer=auto_lock_renewer,
prefetch_count=prefetch_count,
**kwargs
)
self._handlers.add(handler)
Expand Down Expand Up @@ -359,8 +433,21 @@ def get_topic_sender(self, topic_name, **kwargs):
self._handlers.add(handler)
return handler

def get_subscription_receiver(self, topic_name, subscription_name, **kwargs):
# type: (str, str, Any) -> ServiceBusReceiver
def get_subscription_receiver(
self,
topic_name: str,
subscription_name: str,
*,
session_id: Optional[Union[str, ServiceBusSessionFilter]] = None,
sub_queue: Optional[Union[ServiceBusSubQueue, str]] = None,
receive_mode: Union[
ServiceBusReceiveMode, str
] = ServiceBusReceiveMode.PEEK_LOCK,
max_wait_time: Optional[float] = None,
auto_lock_renewer: Optional[AutoLockRenewer] = None,
prefetch_count: int = 0,
**kwargs: Any
) -> ServiceBusReceiver:
"""Get ServiceBusReceiver for the specific subscription under the topic.

:param str topic_name: The name of specific Service Bus Topic the client connects to.
Expand All @@ -369,7 +456,7 @@ def get_subscription_receiver(self, topic_name, subscription_name, **kwargs):
:keyword session_id: A specific session from which to receive. This must be specified for a
sessionful subscription, otherwise it must be None. In order to receive messages from the next available
session, set this to ~azure.servicebus.NEXT_AVAILABLE_SESSION.
:paramtype session_id: Union[str, ~azure.servicebus.NEXT_AVAILABLE_SESSION]
:paramtype session_id: Optional[Union[str, ~azure.servicebus.NEXT_AVAILABLE_SESSION]]
:keyword Optional[Union[ServiceBusSubQueue, str]] sub_queue: If specified, the subqueue this receiver will
connect to.
This includes the DEAD_LETTER and TRANSFER_DEAD_LETTER queues, holds messages that can't be delivered to any
Expand Down Expand Up @@ -415,8 +502,7 @@ def get_subscription_receiver(self, topic_name, subscription_name, **kwargs):
"the connection string used to construct the ServiceBusClient."
)

sub_queue = kwargs.get("sub_queue", None)
if sub_queue and kwargs.get("session_id"):
if sub_queue and session_id:
raise ValueError(
"session_id and sub_queue can not be specified simultaneously. "
"To connect to the sub queue of a sessionful subscription, "
Expand Down Expand Up @@ -444,6 +530,12 @@ def get_subscription_receiver(self, topic_name, subscription_name, **kwargs):
retry_total=self._config.retry_total,
retry_backoff_factor=self._config.retry_backoff_factor,
retry_backoff_max=self._config.retry_backoff_max,
session_id=session_id,
sub_queue=sub_queue,
receive_mode=receive_mode,
max_wait_time=max_wait_time,
auto_lock_renewer=auto_lock_renewer,
prefetch_count=prefetch_count,
**kwargs
)
except ValueError:
Expand All @@ -465,6 +557,12 @@ def get_subscription_receiver(self, topic_name, subscription_name, **kwargs):
retry_total=self._config.retry_total,
retry_backoff_factor=self._config.retry_backoff_factor,
retry_backoff_max=self._config.retry_backoff_max,
session_id=session_id,
sub_queue=sub_queue,
receive_mode=receive_mode,
max_wait_time=max_wait_time,
auto_lock_renewer=auto_lock_renewer,
prefetch_count=prefetch_count,
**kwargs
)
self._handlers.add(handler)
Expand Down
Loading