Fix linting issues in the Azure Remote Rendering SDK (#35517)

* Fix a lot of linting issues

* Fix mypy issues

* Remove type from kwargs

* Remove any added kwargs: Any instance

---------

Co-authored-by: Michael Zappe <[email protected]>

Author: MichaelZp0

sdk/remoterendering/azure-mixedreality-remoterendering/azure/mixedreality/remoterendering/_polling.py

@@ -6,87 +6,73 @@
 import base64
 import time
 from functools import partial
-from typing import TYPE_CHECKING
+from typing import Any, Callable, Union
 
 from azure.core.polling import PollingMethod
 from azure.core.exceptions import HttpResponseError, ODataV4Format
 
 from ._generated.models import (AssetConversion, AssetConversionStatus,
                                 RenderingSession, RenderingSessionStatus)
 
-# pylint: disable=unsubscriptable-object
-if TYPE_CHECKING:
-    # pylint: disable=unused-import,ungrouped-imports
-    from typing import Any, Callable, Union
-
-    from azure.core.credentials import TokenCredential
-
-    from ._generated import RemoteRenderingRestClient
+from ._generated import RemoteRenderingRestClient
 
 
 class RemoteRenderingPolling(PollingMethod):
     """ Abstract base class for polling.
     """
 
-    def __init__(self, account_id, is_terminated, polling_interval=5):
-        # type: (str, Callable, int) -> None
+    def __init__(self, account_id: str, is_terminated: Callable, polling_interval: int=5) -> None:
         self._account_id = account_id
-        self._response = None  # type: Union[AssetConversion, RenderingSession, None]
-        self._client = None  # type: Union[RemoteRenderingRestClient, None]
+        self._response: Union[AssetConversion, RenderingSession, None] = None
+        self._client: Union[RemoteRenderingRestClient, None] = None
         self._query_status = None
         self._is_terminated = is_terminated
         self._polling_interval = polling_interval
 
-    def _update_status(self):
-        # type: () -> None
+    def _update_status(self) -> None:
         if self._query_status is None:
             raise RuntimeError("this poller has not been initialized")
         self._response = self._query_status()  # pylint: disable=E1102
-        if self._response.error is not None:
+        if self._response.error is not None: # type: ignore
             error = HttpResponseError("Polling returned a status indicating an error state.", model=self._response)
-            error.error = ODataV4Format(self._response.error.serialize())
+            error.error = ODataV4Format(self._response.error.serialize()) # type: ignore
             raise error
 
-    def initialize(self, client, initial_response, deserialization_callback):
-        # type: (Any, Any, Callable) -> None
+    def initialize(self, client: Any, initial_response: Any, deserialization_callback: Callable) -> None:
         self._client = client
         self._response = initial_response
 
-    def run(self):
-        # type: () -> None
+    def run(self) -> None:
         while not self.finished():
             self._update_status()
             if not self.finished():
                 time.sleep(self._polling_interval)
 
-    def status(self):
-        # type: () -> str
+    def status(self) -> str:
         raise NotImplementedError("This method needs to be implemented")
 
-    def finished(self):
-        # type: () -> bool
+    def finished(self) -> bool:
         if self._response is None:
             return False
         if self._response.status is None:
             return False
         return self._is_terminated(self._response.status)
 
-    def resource(self):
-        # type: () -> Union[AssetConversion, RenderingSession, None]
+    def resource(self) -> Union[AssetConversion, RenderingSession, None]:
         if not self.finished():
             return None
         return self._response
 
-    def get_continuation_token(self):
-        # type() -> str
-
+    def get_continuation_token(self) -> str:
         # returns a Base64 encoded string of "<version>:<account_id>:<session_id/conversion_id>"
-        return base64.b64encode(("1:"+self._account_id+":"+self._response.id).encode('ascii')).decode('ascii')
+        token_str = "1:"+self._account_id+":"+self._response.id # type: ignore
+        encoded = token_str.encode('ascii')
+        base64_endcoded = base64.b64encode(encoded)
+        return base64_endcoded.decode('ascii')
 
 
 class ConversionPolling(RemoteRenderingPolling):
-    def __init__(self, account_id, polling_interval=5):
-        # type: (str, int) -> None
+    def __init__(self, account_id: str, polling_interval: int=5) -> None:
         def is_terminated(status):
             return status in [
                 AssetConversionStatus.FAILED,
@@ -97,26 +83,23 @@ def is_terminated(status):
                                                 is_terminated=is_terminated,
                                                 polling_interval=polling_interval)
 
-    def initialize(self, client, initial_response, deserialization_callback):
-        # type: (Any, Any, Any, Callable) -> None
+    def initialize(self, client: Any, initial_response: Any, deserialization_callback: Callable) -> None:
         super(ConversionPolling, self).initialize(client=client,
                                                   initial_response=initial_response,
                                                   deserialization_callback=deserialization_callback)
         self._query_status = partial(
-            self._client.remote_rendering.get_conversion,
+            self._client.remote_rendering.get_conversion, # type: ignore
             account_id=self._account_id,
-            conversion_id=initial_response.id)
+            conversion_id=initial_response.id) # type: ignore
 
-    def status(self):
-        # type: () -> str
+    def status(self) -> str:
         if self._response is None:
             return AssetConversionStatus.NOT_STARTED
         return self._response.status
 
     @classmethod
     def from_continuation_token(cls, continuation_token, client, **kwargs):  # pylint: disable=W0221
         # type(str, RemoteRenderingRestClient, Any) -> Tuple
-
         version, account_id, conversion_id = base64.b64decode(
             continuation_token.encode('ascii')).decode('ascii').split(":")
 
@@ -132,8 +115,7 @@ def from_continuation_token(cls, continuation_token, client, **kwargs):  # pylin
 
 
 class SessionPolling(RemoteRenderingPolling):
-    def __init__(self, account_id, polling_interval=2):
-        # type: (str, int) -> None
+    def __init__(self, account_id: str, polling_interval: int=2) -> None:
         def is_terminated(status):
             return status in [
                 RenderingSessionStatus.EXPIRED,
@@ -145,22 +127,21 @@ def is_terminated(status):
                                              is_terminated=is_terminated,
                                              polling_interval=polling_interval)
 
-    def initialize(self, client, initial_response, deserialization_callback):
-        # type: (Any, Any, Any, Callable) -> None
+    def initialize(self, client: Any, initial_response: Any, deserialization_callback: Callable) -> None:
         super(SessionPolling, self).initialize(client, initial_response, deserialization_callback)
         self._query_status = partial(
-            self._client.remote_rendering.get_session, account_id=self._account_id, session_id=initial_response.id)
+            self._client.remote_rendering.get_session, # type: ignore
+            account_id=self._account_id,
+            session_id=initial_response.id) # type: ignore
 
-    def status(self):
-        # type: () -> str
+    def status(self) -> str:
         if self._response is None:
             return RenderingSessionStatus.STARTING
         return self._response.status
 
     @classmethod
     def from_continuation_token(cls, continuation_token, client, **kwargs):  # pylint: disable=W0221
         # type(str, RemoteRenderingRestClient, Any) -> Tuple
-
         version, account_id, session_id = base64.b64decode(
             continuation_token.encode('ascii')).decode('ascii').split(":")
 

sdk/remoterendering/azure-mixedreality-remoterendering/azure/mixedreality/remoterendering/_remote_rendering_client.py

@@ -6,13 +6,14 @@
 # --------------------------------------------------------------------------
 
 
-from typing import TYPE_CHECKING, Any, Callable, Union
+from typing import Any, Callable, Union
 
 from azure.core.credentials import AccessToken, AzureKeyCredential
 from azure.core.paging import ItemPaged
 from azure.core.pipeline.policies import BearerTokenCredentialPolicy
 from azure.core.polling import LROPoller
 from azure.core.tracing.decorator import distributed_trace
+from azure.core.credentials import TokenCredential
 
 from ._api_version import validate_api_version, DEFAULT_VERSION
 from ._generated import RemoteRenderingRestClient
@@ -32,12 +33,6 @@
 from ._shared.static_access_token_credential import StaticAccessTokenCredential
 from ._version import SDK_MONIKER
 
-if TYPE_CHECKING:
-    from azure.core.credentials import TokenCredential
-
-# pylint: disable=unsubscriptable-object
-
-
 class RemoteRenderingClient:
     """A client for the Azure Remote Rendering Service.
 
@@ -126,8 +121,11 @@ def __init__(self,
             **kwargs)
 
     @distributed_trace
-    def begin_asset_conversion(self, conversion_id, input_settings, output_settings, **kwargs):
-        # type: (str, AssetConversionInputSettings, AssetConversionOutputSettings, Any) -> LROPoller[AssetConversion]
+    def begin_asset_conversion(self,
+                               conversion_id: str,
+                               input_settings: AssetConversionInputSettings,
+                               output_settings: AssetConversionOutputSettings,
+                               **kwargs) -> LROPoller[AssetConversion]:
         """
         Start a new asset conversion with the given options.
         :param str conversion_id:
@@ -161,8 +159,7 @@ def begin_asset_conversion(self, conversion_id, input_settings, output_settings,
                          polling_method=polling_method)
 
     @distributed_trace
-    def get_asset_conversion(self, conversion_id, **kwargs):
-        # type: (str, Any) -> AssetConversion
+    def get_asset_conversion(self, conversion_id: str, **kwargs) -> AssetConversion:
         """
         Retrieve the state of a previously created conversion.
         :param str conversion_id:
@@ -175,23 +172,19 @@ def get_asset_conversion(self, conversion_id, **kwargs):
             account_id=self._account_id, conversion_id=conversion_id, **kwargs)
 
     @distributed_trace
-    def get_asset_conversion_poller(self, **kwargs):
-        # type: (Any) -> LROPoller[AssetConversion]
+    def get_asset_conversion_poller(self, **kwargs) -> LROPoller[AssetConversion]: # pylint:disable=docstring-keyword-should-match-keyword-only
         """
         Returns a poller for an existing conversion by conversion id or a continuation token retrieved from a previous
         poller.
-        :keyword conversion_id: The conversion id of a previously created conversion.
-        :paramtype conversion_id: str
-        :keyword continuation_token:
-            A continuation token retrieved from a poller of a conversion.
-        :paramtype continuation_token: str
+        :keyword str conversion_id: The conversion_id of a previously created conversion.
+        :keyword str continuation_token: A continuation token retrieved from a poller of a conversion.
         :return: A poller for the created asset conversion
         :rtype: ~azure.core.polling.LROPoller[AssetConversion]
         :raises ~azure.core.exceptions.HttpResponseError:
         """
 
-        conversion_id = kwargs.pop("conversion_id", None)  # type: Union[str,None]
-        continuation_token = kwargs.pop("continuation_token", None)  # type: Union[str,None]
+        conversion_id: Union[str,None] = kwargs.pop("conversion_id", None)
+        continuation_token: Union[str,None] = kwargs.pop("continuation_token", None)
 
         if conversion_id is None and continuation_token is None:
             raise ValueError(
@@ -222,18 +215,20 @@ def get_asset_conversion_poller(self, **kwargs):
                          polling_method=polling_method)
 
     @distributed_trace
-    def list_asset_conversions(self, **kwargs):
-        # type: (...) -> ItemPaged[AssetConversion]
+    def list_asset_conversions(self, **kwargs) -> ItemPaged[AssetConversion]:
         """ Returns list of conversions for the remote rendering account.
         :rtype: ItemPaged[AssetConversion]
         :raises ~azure.core.exceptions.HttpResponseError:
         :return: List of conversion for the remote rendering account.
         """
-        return self._client.remote_rendering.list_conversions(account_id=self._account_id, **kwargs)  # type: ignore
+        return self._client.remote_rendering.list_conversions(account_id=self._account_id, **kwargs) # type: ignore
 
     @distributed_trace
-    def begin_rendering_session(self, session_id, size, lease_time_minutes, **kwargs):
-        # type: (str, Union[str, RenderingSessionSize], int, Any) -> LROPoller[RenderingSession]
+    def begin_rendering_session(self,
+                                session_id: str,
+                                size: Union[str, RenderingSessionSize],
+                                lease_time_minutes: int,
+                                **kwargs) -> LROPoller[RenderingSession]:
         """
         :param str session_id: An ID uniquely identifying the rendering session for the given account. The ID is case
             sensitive, can contain any combination of alphanumeric characters including hyphens and underscores, and
@@ -263,8 +258,7 @@ def begin_rendering_session(self, session_id, size, lease_time_minutes, **kwargs
                          polling_method=polling_method)
 
     @distributed_trace
-    def get_rendering_session(self, session_id, **kwargs):
-        # type: (str, Any) -> RenderingSession
+    def get_rendering_session(self, session_id: str, **kwargs) -> RenderingSession:
         '''
         Returns the properties of a previously generated rendering session.
         :param str session_id: The identifier of the rendering session.
@@ -278,23 +272,19 @@ def get_rendering_session(self, session_id, **kwargs):
             **kwargs)
 
     @distributed_trace
-    def get_rendering_session_poller(self, **kwargs):
-        # type: (Any) -> LROPoller[RenderingSession]
+    def get_rendering_session_poller(self, **kwargs) -> LROPoller[RenderingSession]: # pylint:disable=docstring-keyword-should-match-keyword-only
         """
         Returns a poller for an existing rendering session by session id or a continuation token retrieved from a
         previous poller.
-        :keyword session_id: The conversion id of a previously created conversion.
-        :paramtype session_id: str
-        :keyword continuation_token:
-            A continuation token retrieved from a poller of a session.
-        :paramtype continuation_token: str
+        :keyword str session_id: The conversion id of a previously created conversion.
+        :keyword str continuation_token: A continuation token retrieved from a poller of a session.
         :return: A session poller for the given session
         :rtype: LROPoller[RenderingSession]
         :raises ~azure.core.exceptions.HttpResponseError:
         """
 
-        session_id = kwargs.pop("session_id", None)  # type: Union[str,None]
-        continuation_token = kwargs.pop("continuation_token", None)  # type: Union[str,None]
+        session_id: Union[str,None] = kwargs.pop("session_id", None)
+        continuation_token: Union[str,None] = kwargs.pop("continuation_token", None)
 
         if session_id is None and continuation_token is None:
             raise ValueError(
@@ -325,8 +315,7 @@ def get_rendering_session_poller(self, **kwargs):
                          polling_method=polling_method)
 
     @distributed_trace
-    def stop_rendering_session(self, session_id, **kwargs):
-        # type:  (str, Any) -> None
+    def stop_rendering_session(self, session_id: str, **kwargs) -> None:
         """
         :param str session_id: The identifier of the session to be stopped.
         :return: None
@@ -337,20 +326,18 @@ def stop_rendering_session(self, session_id, **kwargs):
             account_id=self._account_id, session_id=session_id, **kwargs)
 
     @distributed_trace
-    def update_rendering_session(self, session_id, **kwargs):
-        # type: (str, Any) -> RenderingSession
+    def update_rendering_session(self, session_id: str, **kwargs) -> RenderingSession: # pylint:disable=docstring-keyword-should-match-keyword-only
         """
         Updates an already existing rendering session.
         :param str session_id: The identifier of the session to be updated.
-        :keyword lease_time_minutes: The new lease time of the rendering session. Has to be strictly larger than
+        :keyword int lease_time_minutes: The new lease time of the rendering session. Has to be strictly larger than
             the previous lease time.
-        :paramtype lease_time_minutes: int
         :return: The properties of the updated session
         :rtype: ~azure.mixedreality.remoterendering.models.RenderingSession
         :raises ~azure.core.exceptions.HttpResponseError:
         """
 
-        lease_time_minutes = kwargs.pop("lease_time_minutes", None)  # type: Union[int,None]
+        lease_time_minutes: Union[int,None] = kwargs.pop("lease_time_minutes", None)
         if lease_time_minutes is not None:
             return self._client.remote_rendering.update_session(account_id=self._account_id,
                                                                 session_id=session_id,
@@ -364,8 +351,7 @@ def update_rendering_session(self, session_id, **kwargs):
                                                          **kwargs)
 
     @distributed_trace
-    def list_rendering_sessions(self, **kwargs):
-        # type: (...) -> ItemPaged[RenderingSession]
+    def list_rendering_sessions(self, **kwargs) -> ItemPaged[RenderingSession]:
         """
         Returns list of rendering sessions in the 'Ready' or 'Starting' state.
         Does not return stopped or failed rendering sessions.
@@ -375,15 +361,12 @@ def list_rendering_sessions(self, **kwargs):
         """
         return self._client.remote_rendering.list_sessions(account_id=self._account_id, **kwargs)  # type: ignore
 
-    def close(self):
-        # type: () -> None
+    def close(self) -> None:
         self._client.close()
 
-    def __enter__(self):
-        # type: () -> RemoteRenderingClient
+    def __enter__(self) -> "RemoteRenderingClient":
         self._client.__enter__()  # pylint:disable=no-member
         return self
 
-    def __exit__(self, *args):
-        # type: (*Any) -> None
+    def __exit__(self, *args: Any) -> None:
         self._client.__exit__(*args)  # pylint:disable=no-member