[Service Bus] Deprecate Internal _Accesstoken

sdk/servicebus/azure-servicebus/azure/servicebus/_base_handler.py

@@ -2,7 +2,6 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
-import collections
 import logging
 import uuid
 import time
@@ -39,7 +38,6 @@
 if TYPE_CHECKING:
     from azure.core.credentials import TokenCredential
 
-_AccessToken = collections.namedtuple("AccessToken", "token expires_on")
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -95,7 +93,7 @@ def _parse_conn_str(conn_str):
 
 
 def _generate_sas_token(uri, policy, key, expiry=None):
-    # type: (str, str, str, Optional[timedelta]) -> _AccessToken
+    # type: (str, str, str, Optional[timedelta]) -> AccessToken
     """Create a shared access signiture token as a string literal.
     :returns: SAS token as string literal.
     :rtype: str
@@ -109,7 +107,7 @@ def _generate_sas_token(uri, policy, key, expiry=None):
     encoded_key = key.encode("utf-8")
 
     token = utils.create_sas_token(encoded_policy, encoded_key, encoded_uri, expiry)
-    return _AccessToken(token=token, expires_on=abs_expiry)
+    return AccessToken(token=token, expires_on=abs_expiry)
 
 
 class ServiceBusSASTokenCredential(object):
@@ -149,7 +147,7 @@ def __init__(self, policy, key):
         self.token_type = TOKEN_TYPE_SASTOKEN
 
     def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
-        # type: (str, Any) -> _AccessToken
+        # type: (str, Any) -> AccessToken
         if not scopes:
             raise ValueError("No token scope provided.")
         return _generate_sas_token(scopes[0], self.policy, self.key)

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_base_handler_async.py

@@ -14,7 +14,7 @@
 
 from azure.core.credentials import AccessToken
 
-from .._base_handler import _generate_sas_token, _AccessToken, BaseHandler as BaseHandlerSync
+from .._base_handler import _generate_sas_token, BaseHandler as BaseHandlerSync
 from .._common._configuration import Configuration
 from .._common.utils import create_properties
 from .._common.constants import (
@@ -67,7 +67,7 @@ def __init__(self, policy: str, key: str) -> None:
         self.key = key
         self.token_type = TOKEN_TYPE_SASTOKEN
 
-    async def get_token(self, *scopes: str, **kwargs: Any) -> _AccessToken:  # pylint:disable=unused-argument
+    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:  # pylint:disable=unused-argument
         if not scopes:
             raise ValueError("No token scope provided.")
         return _generate_sas_token(scopes[0], self.policy, self.key)

sdk/servicebus/azure-servicebus/azure/servicebus/exceptions.py

@@ -4,7 +4,7 @@
 # license information.
 # -------------------------------------------------------------------------
 
-from typing import Optional, Any
+from typing import Optional
 
 from uamqp import errors, constants
 from azure.core.exceptions import AzureError