Custom NoPolling implementations for return types

Author: mccoyp

sdk/keyvault/azure-keyvault-securitydomain/azure/keyvault/securitydomain/_internal/__init__.py

@@ -6,27 +6,38 @@
 
 from urllib.parse import urlparse
 
-from .async_polling import AsyncSecurityDomainDownloadPollingMethod, AsyncSecurityDomainUploadPollingMethod
+from .async_polling import (
+    AsyncSecurityDomainDownloadNoPolling,
+    AsyncSecurityDomainDownloadPollingMethod,
+    AsyncSecurityDomainUploadNoPolling,
+    AsyncSecurityDomainUploadPollingMethod,
+)
 from .challenge_auth_policy import ChallengeAuthPolicy
 from .http_challenge import HttpChallenge
 from . import http_challenge_cache
 from .polling import (
+    SecurityDomainDownloadNoPolling,
     SecurityDomainDownloadPolling,
     SecurityDomainDownloadPollingMethod,
+    SecurityDomainUploadNoPolling,
     SecurityDomainUploadPolling,
     SecurityDomainUploadPollingMethod,
 )
 
 HttpChallengeCache = http_challenge_cache  # to avoid aliasing pylint error (C4745)
 
 __all__ = [
+    "AsyncSecurityDomainDownloadNoPolling",
     "AsyncSecurityDomainDownloadPollingMethod",
+    "AsyncSecurityDomainUploadNoPolling",
     "AsyncSecurityDomainUploadPollingMethod",
     "ChallengeAuthPolicy",
     "HttpChallenge",
     "HttpChallengeCache",
+    "SecurityDomainDownloadNoPolling",
     "SecurityDomainDownloadPolling",
     "SecurityDomainDownloadPollingMethod",
+    "SecurityDomainUploadNoPolling",
     "SecurityDomainUploadPolling",
     "SecurityDomainUploadPollingMethod",
 ]

sdk/keyvault/azure-keyvault-securitydomain/azure/keyvault/securitydomain/_internal/async_polling.py

@@ -4,12 +4,25 @@
 # ------------------------------------
 from azure.core.polling.async_base_polling import AsyncLROBasePolling
 
-from .polling import SecurityDomainDownloadPollingMethod, SecurityDomainUploadPollingMethod
+from .polling import (
+    SecurityDomainDownloadPollingMethod,
+    SecurityDomainDownloadNoPolling,
+    SecurityDomainUploadPollingMethod,
+    SecurityDomainUploadNoPolling,
+)
 
 
 class AsyncSecurityDomainDownloadPollingMethod(AsyncLROBasePolling, SecurityDomainDownloadPollingMethod):
     pass
 
 
+class AsyncSecurityDomainDownloadNoPolling(AsyncLROBasePolling, SecurityDomainDownloadNoPolling):
+    pass
+
+
 class AsyncSecurityDomainUploadPollingMethod(AsyncLROBasePolling, SecurityDomainUploadPollingMethod):
     pass
+
+
+class AsyncSecurityDomainUploadNoPolling(AsyncLROBasePolling, SecurityDomainUploadNoPolling):
+    pass

sdk/keyvault/azure-keyvault-securitydomain/azure/keyvault/securitydomain/_internal/polling.py

@@ -16,8 +16,8 @@
 
 PollingReturnType_co = TypeVar("PollingReturnType_co", covariant=True)
 
-# The correct success response should be "Succeeded", but this has already shipped.
-_FINISHED = frozenset(["success", "canceled", "failed"])
+# The correct success response should be "Succeeded", but this has already shipped. Handle "Succeeded" just in case.
+_FINISHED = frozenset(["succeeded", "success", "canceled", "failed"])
 
 
 def _finished(status):
@@ -71,6 +71,27 @@ def parse_resource(
         return None  # type: ignore
 
 
+class NoPollingMixin(LROBasePolling):
+    def finished(self) -> bool:
+        """Is this polling finished?
+
+        :rtype: bool
+        :return: Whether this polling is finished
+        """
+        return True
+
+    def status(self) -> str:
+        """Return the current status.
+
+        :rtype: str
+        :return: The current status
+        """
+        return "succeeded"
+
+    def result(self, *args, **kwargs):
+        return self.resource()
+
+
 class SecurityDomainDownloadPolling(OperationResourcePolling):
     def __init__(self) -> None:
         self._polling_url = ""
@@ -121,7 +142,9 @@ def get_long_running_output(pipeline_response):
             response_headers["Azure-AsyncOperation"] = deserializer._deserialize(  # pylint: disable=protected-access
                 "str", response.headers.get("Azure-AsyncOperation")
             )
-            response_headers["Retry-After"] = deserializer._deserialize("int", response.headers.get("Retry-After"))  # pylint: disable=protected-access
+            response_headers["Retry-After"] = deserializer._deserialize(
+                "int", response.headers.get("Retry-After")
+            )  # pylint: disable=protected-access
 
             return _deserialize(SecurityDomainObject, response.json())
 
@@ -137,6 +160,10 @@ def resource(self) -> SecurityDomainObject:
         return cast(SecurityDomainObject, self.parse_resource(self._initial_response))
 
 
+class SecurityDomainDownloadNoPolling(SecurityDomainDownloadPollingMethod, NoPollingMixin):
+    pass
+
+
 class SecurityDomainUploadPolling(SecurityDomainDownloadPolling):
     def set_initial_status(self, pipeline_response: "PipelineResponse") -> str:
         response: HttpResponse = pipeline_response.http_response
@@ -175,7 +202,9 @@ def get_long_running_output(pipeline_response):
             response_headers["Azure-AsyncOperation"] = deserializer._deserialize(  # pylint: disable=protected-access
                 "str", response.headers.get("Azure-AsyncOperation")
             )
-            response_headers["Retry-After"] = deserializer._deserialize("int", response.headers.get("Retry-After"))  # pylint: disable=protected-access
+            response_headers["Retry-After"] = deserializer._deserialize(
+                "int", response.headers.get("Retry-After")
+            )  # pylint: disable=protected-access
 
             return _deserialize(SecurityDomainOperationStatus, response.json())
 
@@ -188,3 +217,13 @@ def resource(self) -> SecurityDomainOperationStatus:
         :return: The built resource.
         """
         return cast(SecurityDomainOperationStatus, self.parse_resource(self._pipeline_response))
+
+
+class SecurityDomainUploadNoPolling(SecurityDomainUploadPollingMethod, NoPollingMixin):
+    def resource(self) -> SecurityDomainOperationStatus:
+        """Return the built resource.
+
+        :rtype: any
+        :return: The built resource.
+        """
+        return cast(SecurityDomainOperationStatus, self.parse_resource(self._initial_response))

sdk/keyvault/azure-keyvault-securitydomain/azure/keyvault/securitydomain/_patch.py

@@ -21,8 +21,10 @@
 from ._client import KeyVaultClient
 from ._internal import (
     ChallengeAuthPolicy,
+    SecurityDomainDownloadNoPolling,
     SecurityDomainDownloadPolling,
     SecurityDomainDownloadPollingMethod,
+    SecurityDomainUploadNoPolling,
     SecurityDomainUploadPolling,
     SecurityDomainUploadPollingMethod,
 )
@@ -124,8 +126,7 @@ def begin_download(
         content_type: str = "application/json",
         polling: Optional[Literal[False]] = None,
         **kwargs: Any,
-    ) -> LROPoller[SecurityDomainObject]:
-        ...
+    ) -> LROPoller[SecurityDomainObject]: ...
 
     @overload
     def begin_download(
@@ -135,8 +136,7 @@ def begin_download(
         content_type: str = "application/json",
         polling: Optional[Literal[False]] = None,
         **kwargs: Any,
-    ) -> LROPoller[SecurityDomainObject]:
-        ...
+    ) -> LROPoller[SecurityDomainObject]: ...
 
     @overload
     def begin_download(
@@ -146,8 +146,7 @@ def begin_download(
         content_type: str = "application/json",
         polling: Optional[Literal[False]] = None,
         **kwargs: Any,
-    ) -> LROPoller[SecurityDomainObject]:
-        ...
+    ) -> LROPoller[SecurityDomainObject]: ...
 
     @distributed_trace
     def begin_download(
@@ -178,8 +177,10 @@ def begin_download(
         :raises ~azure.core.exceptions.HttpResponseError:
         """
         delay = kwargs.pop("polling_interval", self._config.polling_interval)
-        polling_method = False if polling is False else SecurityDomainDownloadPollingMethod(
-            lro_algorithms=[SecurityDomainDownloadPolling()], timeout=delay
+        polling_method = (
+            SecurityDomainDownloadNoPolling()
+            if polling is False
+            else SecurityDomainDownloadPollingMethod(lro_algorithms=[SecurityDomainDownloadPolling()], timeout=delay)
         )
         return super().begin_download(  # type: ignore[return-value]
             certificate_info_object,
@@ -215,8 +216,10 @@ def begin_upload(
         :raises ~azure.core.exceptions.HttpResponseError:
         """
         delay = kwargs.pop("polling_interval", self._config.polling_interval)
-        polling_method = False if polling is False else SecurityDomainUploadPollingMethod(
-            lro_algorithms=[SecurityDomainUploadPolling()], timeout=delay
+        polling_method = (
+            SecurityDomainUploadNoPolling()
+            if polling is False
+            else SecurityDomainUploadPollingMethod(lro_algorithms=[SecurityDomainUploadPolling()], timeout=delay)
         )
         return super().begin_upload(
             security_domain,

sdk/keyvault/azure-keyvault-securitydomain/azure/keyvault/securitydomain/aio/_patch.py

@@ -18,7 +18,9 @@
 from ._client import KeyVaultClient
 from .._internal import (
     AsyncChallengeAuthPolicy,
+    AsyncSecurityDomainDownloadNoPolling,
     AsyncSecurityDomainDownloadPollingMethod,
+    AsyncSecurityDomainUploadNoPolling,
     AsyncSecurityDomainUploadPollingMethod,
     SecurityDomainDownloadPolling,
     SecurityDomainUploadPolling,
@@ -85,8 +87,7 @@ async def begin_download(
         content_type: str = "application/json",
         polling: Optional[Literal[False]] = None,
         **kwargs: Any,
-    ) -> AsyncLROPoller[SecurityDomainObject]:
-        ...
+    ) -> AsyncLROPoller[SecurityDomainObject]: ...
 
     @overload
     async def begin_download(
@@ -96,8 +97,7 @@ async def begin_download(
         content_type: str = "application/json",
         polling: Optional[Literal[False]] = None,
         **kwargs: Any,
-    ) -> AsyncLROPoller[SecurityDomainObject]:
-        ...
+    ) -> AsyncLROPoller[SecurityDomainObject]: ...
 
     @overload
     async def begin_download(
@@ -107,8 +107,7 @@ async def begin_download(
         content_type: str = "application/json",
         polling: Optional[Literal[False]] = None,
         **kwargs: Any,
-    ) -> AsyncLROPoller[SecurityDomainObject]:
-        ...
+    ) -> AsyncLROPoller[SecurityDomainObject]: ...
 
     @distributed_trace_async
     async def begin_download(
@@ -139,8 +138,12 @@ async def begin_download(
         :raises ~azure.core.exceptions.HttpResponseError:
         """
         delay = kwargs.pop("polling_interval", self._config.polling_interval)
-        polling_method = False if polling is False else AsyncSecurityDomainDownloadPollingMethod(
-            lro_algorithms=[SecurityDomainDownloadPolling()], timeout=delay
+        polling_method = (
+            AsyncSecurityDomainDownloadNoPolling()
+            if polling is False
+            else AsyncSecurityDomainDownloadPollingMethod(
+                lro_algorithms=[SecurityDomainDownloadPolling()], timeout=delay
+            )
         )
         return await super().begin_download(  # type: ignore[return-value]
             certificate_info_object,
@@ -176,8 +179,10 @@ async def begin_upload(
         :raises ~azure.core.exceptions.HttpResponseError:
         """
         delay = kwargs.pop("polling_interval", self._config.polling_interval)
-        polling_method = False if polling is False else AsyncSecurityDomainUploadPollingMethod(
-            lro_algorithms=[SecurityDomainUploadPolling()], timeout=delay
+        polling_method = (
+            AsyncSecurityDomainUploadNoPolling()
+            if polling is False
+            else AsyncSecurityDomainUploadPollingMethod(lro_algorithms=[SecurityDomainUploadPolling()], timeout=delay)
         )
         return await super().begin_upload(
             security_domain,