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 1 commit
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
128 changes: 111 additions & 17 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,29 +2,39 @@
# 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
import logging
from weakref import WeakSet

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

from azure.core.credentials import (
TokenCredential,
AzureSasCredential,
AzureNamedKeyCredential,
)
from ._base_handler import (
_parse_conn_str,
ServiceBusSharedKeyCredential,
ServiceBusSASTokenCredential,
)
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,
NEXT_AVAILABLE_SESSION,
ServiceBusReceiveMode,
)
from ._retry import RetryMode

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

_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -74,15 +84,38 @@ 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 +166,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
):
"""
Create a ServiceBusClient from a connection string.

Expand Down Expand Up @@ -179,6 +224,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,8 +278,20 @@ 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: Union[str, NEXT_AVAILABLE_SESSION] = None,

@swathipil swathipil Jan 6, 2022 •
edited
Loading

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.

@yunhaoling @rakshith91
why is NEXT_AVAILABLE_SESSION available in the public API, rather than ServiceBusSessionFilter. I know it's the only option for now, but it seems strange.

If we only want that constant to be available, rather than any other future values added to ServiceBusSessionFilter, should the type be Union[str, int]? And, since None is an option, shouldn't the type here be Optional[Union[str, int]]?

@yunhaoling yunhaoling Jan 6, 2022

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.

you're right... it's a bit strange here as NEXT_AVAILABLE_SESSION is a constant value not a class. Also, ServiceBusSessionFilter is not an extensible Enum so it's not of type int, Union[str, int] wouldn't work unless we make it an extensible Enum -- class ServiceBusSessionFilter(int, Enum).

I feel that the right approach is to expose ServiceBusSessionFilter, however, I feel it's overkilling to expose an Enum with just one member.

how about let's keep it for now unless someone explains?

@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.

I'm def with you on that. Unfortunately, this Enum is not a valid type, so I get a mypy error. Maybe I can just try to ignore the type? Not sure if that would work, but is that preferable?

@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.

remove explicit kwarg for now, if resolution not reached

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.
Expand Down Expand Up @@ -278,8 +343,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 +376,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 +429,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: Union[str, NEXT_AVAILABLE_SESSION] = NEXT_AVAILABLE_SESSION,
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 Down Expand Up @@ -415,8 +498,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 +526,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 +553,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