[Communication Shared] Adding the mypy fixes (#42925)

* Adding the mypy fixes

* addressing the comments

* addressing comments

* Make docs happy

* Updated docstring references

---------

Co-authored-by: antisch <[email protected]>

Author: v-dharmarajv

sdk/communication/azure-communication-callautomation/azure/communication/callautomation/_shared/auth_policy_utils.py

@@ -28,11 +28,10 @@ def get_authentication_policy(
     :type credential: Union[TokenCredential, AsyncTokenCredential, AzureKeyCredential, str]
     :param bool decode_url: `True` if there is a need to decode the url. Default value is `False`
     :param bool is_async: For async clients there is a need to decode the url
-
-    :return: Either AsyncBearerTokenCredentialPolicy or BearerTokenCredentialPolicy or HMACCredentialsPolicy
+    :return: The authentication policy to be used.
     :rtype: ~azure.core.pipeline.policies.AsyncBearerTokenCredentialPolicy or
-    ~azure.core.pipeline.policies.BearerTokenCredentialPolicy or
-    ~azure.communication.callautomation.shared.policy.HMACCredentialsPolicy
+     ~azure.core.pipeline.policies.BearerTokenCredentialPolicy or
+     ~.HMACCredentialsPolicy
     """
 
     if credential is None:

sdk/communication/azure-communication-callautomation/azure/communication/callautomation/_shared/models.py

@@ -245,7 +245,7 @@ def __init__(self, user_id: str, **kwargs: Any) -> None:
         :param str user_id: Microsoft Teams user id.
         :keyword bool is_anonymous: `True` if the identifier is anonymous. Default value is `False`.
         :keyword cloud: Cloud environment that the user belongs to. Default value is `PUBLIC`.
-        :paramtype cloud: str or ~azure.communication.callautomation.CommunicationCloudEnvironment
+        :paramtype cloud: str or :class:`~.CommunicationCloudEnvironment`
         :keyword str raw_id: The raw ID of the identifier. If not specified, this value will be constructed from
          the other properties.
         """
@@ -316,7 +316,7 @@ def __init__(self, app_id: str, **kwargs: Any) -> None:
         """
         :param str app_id: Microsoft Teams application id.
         :keyword cloud: Cloud environment that the application belongs to. Default value is `PUBLIC`.
-        :paramtype cloud: str or ~azure.communication.callautomation.CommunicationCloudEnvironment
+        :paramtype cloud: str or :class:`~.CommunicationCloudEnvironment`
         :keyword str raw_id: The raw ID of the identifier. If not specified, this value will be constructed
          from the other properties.
         """
@@ -360,7 +360,7 @@ def __init__(self, bot_id, **kwargs):
         :keyword bool is_resource_account_configured: `False` if the identifier is global.
          Default value is `True` for tennantzed bots.
         :keyword cloud: Cloud environment that the bot belongs to. Default value is `PUBLIC`.
-        :paramtype cloud: str or ~azure.communication.callautomation.CommunicationCloudEnvironment
+        :paramtype cloud: str or :class:`~.CommunicationCloudEnvironment`
         """
         warnings.warn(
             "The MicrosoftBotIdentifier is deprecated and has been replaced by MicrosoftTeamsAppIdentifier.",
@@ -398,7 +398,7 @@ def __init__(self, *, user_id: str, tenant_id: str, resource_id: str, **kwargs:
         :param str tenant_id: Tenant id associated with the user.
         :param str resource_id: The Communication Services resource id.
         :keyword cloud: Cloud environment that the user belongs to. Default value is `PUBLIC`.
-        :paramtype cloud: str or ~azure.communication.callautomation.CommunicationCloudEnvironment
+        :paramtype cloud: str or :class:`~.CommunicationCloudEnvironment`
         :keyword str raw_id: The raw ID of the identifier.
          If not specified, this value will be constructed from the other properties.
         """
@@ -455,7 +455,7 @@ def identifier_from_raw_id(raw_id: str) -> CommunicationIdentifier:  # pylint: d
 
     :param str raw_id: A raw ID to construct the CommunicationIdentifier from.
     :return: The CommunicationIdentifier parsed from the raw_id.
-    :rtype: CommunicationIdentifier
+    :rtype: :class:`~.CommunicationIdentifier`
     """
     if raw_id.startswith(PHONE_NUMBER_PREFIX):
         return PhoneNumberIdentifier(value=raw_id[len(PHONE_NUMBER_PREFIX) :], raw_id=raw_id)

sdk/communication/azure-communication-chat/azure/communication/chat/_shared/auth_policy_utils.py

@@ -4,7 +4,7 @@
 # license information.
 # -------------------------------------------------------------------------
 
-from typing import Union
+from typing import Union, cast
 from azure.core.credentials import TokenCredential, AzureKeyCredential
 from azure.core.credentials_async import AsyncTokenCredential
 from azure.core.pipeline.policies import (
@@ -28,21 +28,22 @@ def get_authentication_policy(
     :type credential: Union[TokenCredential, AsyncTokenCredential, AzureKeyCredential, str]
     :param bool decode_url: `True` if there is a need to decode the url. Default value is `False`
     :param bool is_async: For async clients there is a need to decode the url
-
-    :return: Either AsyncBearerTokenCredentialPolicy or BearerTokenCredentialPolicy or HMACCredentialsPolicy
+    :return: The authentication policy to be used.
     :rtype: ~azure.core.pipeline.policies.AsyncBearerTokenCredentialPolicy or
-    ~azure.core.pipeline.policies.BearerTokenCredentialPolicy or
-    ~azure.communication.chat.shared.policy.HMACCredentialsPolicy
+     ~azure.core.pipeline.policies.BearerTokenCredentialPolicy or
+     ~.HMACCredentialsPolicy
     """
 
     if credential is None:
         raise ValueError("Parameter 'credential' must not be None.")
     if hasattr(credential, "get_token"):
         if is_async:
             return AsyncBearerTokenCredentialPolicy(
-                credential, "https://communication.azure.com//.default"  # type: ignore
+                cast(AsyncTokenCredential, credential), "https://communication.azure.com//.default"
             )
-        return BearerTokenCredentialPolicy(credential, "https://communication.azure.com//.default")  # type: ignore
+        return BearerTokenCredentialPolicy(
+            cast(TokenCredential, credential), "https://communication.azure.com//.default"
+        )
     if isinstance(credential, (AzureKeyCredential, str)):
         return HMACCredentialsPolicy(endpoint, credential, decode_url=decode_url)
 

sdk/communication/azure-communication-chat/azure/communication/chat/_shared/models.py

@@ -157,14 +157,16 @@ def __init__(self, value: str, **kwargs: Any) -> None:
         is_anonymous: bool
 
         if raw_id is not None:
-            phone_number = raw_id[len(PHONE_NUMBER_PREFIX):]
+            phone_number = raw_id[len(PHONE_NUMBER_PREFIX) :]
             is_anonymous = phone_number == PHONE_NUMBER_ANONYMOUS_SUFFIX
             asserted_id_index = -1 if is_anonymous else phone_number.rfind("_") + 1
             has_asserted_id = 0 < asserted_id_index < len(phone_number)
-            props = {"value": value, "is_anonymous": is_anonymous}
             if has_asserted_id:
-                props["asserted_id"] = phone_number[asserted_id_index:]
-            self.properties = PhoneNumberProperties(**props)  # type: ignore
+                self.properties = PhoneNumberProperties(
+                    value=value, is_anonymous=is_anonymous, asserted_id=phone_number[asserted_id_index:]
+                )
+            else:
+                self.properties = PhoneNumberProperties(value=value, is_anonymous=is_anonymous)
         else:
             self.properties = PhoneNumberProperties(value=value)
         self.raw_id = raw_id if raw_id is not None else self._format_raw_id(self.properties)
@@ -183,6 +185,7 @@ def _format_raw_id(self, properties: PhoneNumberProperties) -> str:
         value = properties["value"]
         return f"{PHONE_NUMBER_PREFIX}{value}"
 
+
 class UnknownIdentifier:
     """Represents an identifier of an unknown type.
 
@@ -242,7 +245,7 @@ def __init__(self, user_id: str, **kwargs: Any) -> None:
         :param str user_id: Microsoft Teams user id.
         :keyword bool is_anonymous: `True` if the identifier is anonymous. Default value is `False`.
         :keyword cloud: Cloud environment that the user belongs to. Default value is `PUBLIC`.
-        :paramtype cloud: str or ~azure.communication.chat.CommunicationCloudEnvironment
+        :paramtype cloud: str or :class:`~.CommunicationCloudEnvironment`
         :keyword str raw_id: The raw ID of the identifier. If not specified, this value will be constructed from
          the other properties.
         """
@@ -313,7 +316,7 @@ def __init__(self, app_id: str, **kwargs: Any) -> None:
         """
         :param str app_id: Microsoft Teams application id.
         :keyword cloud: Cloud environment that the application belongs to. Default value is `PUBLIC`.
-        :paramtype cloud: str or ~azure.communication.chat.CommunicationCloudEnvironment
+        :paramtype cloud: str or :class:`~.CommunicationCloudEnvironment`
         :keyword str raw_id: The raw ID of the identifier. If not specified, this value will be constructed
          from the other properties.
         """
@@ -357,7 +360,7 @@ def __init__(self, bot_id, **kwargs):
         :keyword bool is_resource_account_configured: `False` if the identifier is global.
          Default value is `True` for tennantzed bots.
         :keyword cloud: Cloud environment that the bot belongs to. Default value is `PUBLIC`.
-        :paramtype cloud: str or ~azure.communication.chat.CommunicationCloudEnvironment
+        :paramtype cloud: str or :class:`~.CommunicationCloudEnvironment`
         """
         warnings.warn(
             "The MicrosoftBotIdentifier is deprecated and has been replaced by MicrosoftTeamsAppIdentifier.",
@@ -389,20 +392,13 @@ class TeamsExtensionUserIdentifier:
     raw_id: str
     """The raw ID of the identifier."""
 
-    def __init__(
-        self,
-        *,
-        user_id: str,
-        tenant_id: str,
-        resource_id: str,
-        **kwargs: Any
-    ) -> None:
+    def __init__(self, *, user_id: str, tenant_id: str, resource_id: str, **kwargs: Any) -> None:
         """
         :param str user_id: Teams extension user id.
         :param str tenant_id: Tenant id associated with the user.
         :param str resource_id: The Communication Services resource id.
         :keyword cloud: Cloud environment that the user belongs to. Default value is `PUBLIC`.
-        :paramtype cloud: str or ~azure.communication.chat.CommunicationCloudEnvironment
+        :paramtype cloud: str or :class:`~.CommunicationCloudEnvironment`
         :keyword str raw_id: The raw ID of the identifier.
          If not specified, this value will be constructed from the other properties.
         """
@@ -434,6 +430,7 @@ def _format_raw_id(self, properties: TeamsExtensionUserProperties) -> str:
             prefix = ACS_USER_PREFIX
         return f"{prefix}{properties['resource_id']}_{properties['tenant_id']}_{properties['user_id']}"
 
+
 def try_create_teams_extension_user(prefix: str, suffix: str) -> Optional[TeamsExtensionUserIdentifier]:
     segments = suffix.split("_")
     if len(segments) != 3:
@@ -449,6 +446,7 @@ def try_create_teams_extension_user(prefix: str, suffix: str) -> Optional[TeamsE
         raise ValueError("Invalid MRI")
     return TeamsExtensionUserIdentifier(user_id=user_id, tenant_id=tenant_id, resource_id=resource_id, cloud=cloud)
 
+
 def identifier_from_raw_id(raw_id: str) -> CommunicationIdentifier:  # pylint: disable=too-many-return-statements
     """
     Creates a CommunicationIdentifier from a given raw ID.
@@ -457,7 +455,7 @@ def identifier_from_raw_id(raw_id: str) -> CommunicationIdentifier:  # pylint: d
 
     :param str raw_id: A raw ID to construct the CommunicationIdentifier from.
     :return: The CommunicationIdentifier parsed from the raw_id.
-    :rtype: CommunicationIdentifier
+    :rtype: :class:`~.CommunicationIdentifier`
     """
     if raw_id.startswith(PHONE_NUMBER_PREFIX):
         return PhoneNumberIdentifier(value=raw_id[len(PHONE_NUMBER_PREFIX) :], raw_id=raw_id)

sdk/communication/azure-communication-chat/azure/communication/chat/_shared/policy.py

@@ -5,13 +5,15 @@
 # -------------------------------------------------------------------------
 
 import hashlib
-import urllib
 import base64
 import hmac
-from urllib.parse import ParseResult, urlparse
+from urllib.parse import urlparse, unquote
 from typing import Union
+
 from azure.core.credentials import AzureKeyCredential
 from azure.core.pipeline.policies import SansIOHTTPPolicy
+from azure.core.pipeline import PipelineRequest
+
 from .utils import get_current_utc_time
 
 
@@ -41,9 +43,7 @@ def __init__(
         self._access_key = access_key
         self._decode_url = decode_url
 
-    def _compute_hmac(
-        self, value  # type: str
-    ):
+    def _compute_hmac(self, value: str) -> str:
         if isinstance(self._access_key, AzureKeyCredential):
             decoded_secret = base64.b64decode(self._access_key.key)
         else:
@@ -53,11 +53,11 @@ def _compute_hmac(
 
         return base64.b64encode(digest).decode("utf-8")
 
-    def _sign_request(self, request):
+    def _sign_request(self, request: PipelineRequest) -> None:
         verb = request.http_request.method.upper()
 
         # Get the path and query from url, which looks like https://host/path/query
-        parsed_url: ParseResult = urlparse(request.http_request.url)
+        parsed_url = urlparse(request.http_request.url)
         query_url = parsed_url.path
 
         if parsed_url.query:
@@ -91,7 +91,7 @@ def _sign_request(self, request):
             pass
 
         if self._decode_url:
-            query_url = urllib.parse.unquote(query_url)
+            query_url = unquote(query_url)
 
         signed_headers = "x-ms-date;host;x-ms-content-sha256"
 
@@ -114,7 +114,5 @@ def _sign_request(self, request):
 
         request.http_request.headers.update(signature_header)
 
-        return request
-
-    def on_request(self, request):
+    def on_request(self, request: PipelineRequest) -> None:
         self._sign_request(request)

sdk/communication/azure-communication-chat/azure/communication/chat/_shared/token_exchange.py

@@ -6,6 +6,7 @@
 
 import json
 from typing import Any, List, Optional
+
 # pylint: disable=non-abstract-transport-import
 # pylint: disable=no-name-in-module
 
@@ -21,7 +22,7 @@
 
 class TokenExchangeClient:
     """Represents a client that exchanges an Entra token for an Azure Communication Services (ACS) token.
-    
+
     :param resource_endpoint: The endpoint URL of the resource to authenticate against.
     :param credential: The credential to use for token exchange.
     :param scopes: The scopes to request during the token exchange.
@@ -31,11 +32,8 @@ class TokenExchangeClient:
     # pylint: disable=C4748
     # pylint: disable=client-method-missing-type-annotations
     def __init__(
-            self,
-            resource_endpoint: str,
-            credential: TokenCredential,
-            scopes: Optional[List[str]] = None,
-            **kwargs: Any):
+        self, resource_endpoint: str, credential: TokenCredential, scopes: Optional[List[str]] = None, **kwargs: Any
+    ):
 
         self._resource_endpoint = resource_endpoint
         self._scopes = scopes or ["https://communication.azure.com/clients/.default"]
@@ -76,6 +74,5 @@ def _parse_access_token_from_response(self, response: PipelineResponse) -> Acces
                 raise ValueError("Failed to parse access token from response") from ex
         else:
             raise HttpResponseError(
-                message="Failed to exchange Entra token for ACS token",
-                response=response.http_response
+                message="Failed to exchange Entra token for ACS token", response=response.http_response
             )

sdk/communication/azure-communication-chat/azure/communication/chat/_shared/token_exchange_async.py

@@ -6,6 +6,7 @@
 
 import json
 from typing import Any, Optional, List
+
 # pylint: disable=non-abstract-transport-import
 # pylint: disable=no-name-in-module
 
@@ -35,7 +36,8 @@ def __init__(
         resource_endpoint: str,
         credential: AsyncTokenCredential,
         scopes: Optional[List[str]] = None,
-        **kwargs: Any):
+        **kwargs: Any
+    ):
 
         self._resource_endpoint = resource_endpoint
         self._scopes = scopes or ["https://communication.azure.com/clients/.default"]
@@ -76,6 +78,5 @@ async def _parse_access_token_from_response(self, response: PipelineResponse) ->
                 raise ValueError("Failed to parse access token from response") from ex
         else:
             raise HttpResponseError(
-                message="Failed to exchange Entra token for ACS token",
-                response=response.http_response
+                message="Failed to exchange Entra token for ACS token", response=response.http_response
             )

sdk/communication/azure-communication-chat/azure/communication/chat/_shared/token_utils.py

@@ -21,6 +21,7 @@
 
 def create_request_message(resource_endpoint: str, scopes: Optional[List[str]]) -> Any:
     from azure.core.pipeline.transport import HttpRequest
+
     request_uri = create_request_uri(resource_endpoint, scopes)
     request = HttpRequest("POST", request_uri)
     request.headers["Accept"] = "application/json"
@@ -59,33 +60,32 @@ def parse_expires_on(expires_on, response):
             return expires_on_epoch
         except Exception as exc:
             raise HttpResponseError(
-                message="Unknown format for expires_on field in access token response",
-                response=response.http_response) from exc
+                message="Unknown format for expires_on field in access token response", response=response.http_response
+            ) from exc
     else:
         raise HttpResponseError(
-            message="Missing expires_on field in access token response",
-            response=response.http_response)
+            message="Missing expires_on field in access token response", response=response.http_response
+        )
 
 
 def is_entra_token_cache_valid(entra_token_cache, request):
     current_entra_token = request.http_request.headers.get("Authorization", "")
-    cache_valid = (
-        entra_token_cache is not None and
-        current_entra_token == entra_token_cache
-    )
+    cache_valid = entra_token_cache is not None and current_entra_token == entra_token_cache
     return cache_valid, current_entra_token
 
 
 def is_acs_token_cache_valid(response_cache):
-    if (response_cache is None or response_cache.http_response is None or
-            response_cache.http_response.status_code != 200):
+    if (
+        response_cache is None
+        or response_cache.http_response is None
+        or response_cache.http_response.status_code != 200
+    ):
         return False
     try:
         content = response_cache.http_response.text()
         data = json.loads(content)
         expires_on = data["accessToken"]["expiresOn"]
         expires_on_dt = isodate.parse_datetime(expires_on)
         return datetime.now(timezone.utc) < expires_on_dt
-    except (KeyError, ValueError, json.JSONDecodeError):
-        raise ValueError(  # pylint: disable=W0707
-                "Invalid token response")
+    except (KeyError, ValueError, json.JSONDecodeError) as e:
+        raise ValueError("Invalid token response") from e