feat: wrap async rest methods with retry and error logic

gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2

@@ -1,3 +1,4 @@
+{% import "%namespace/%name_%version/%sub/services/%service/_shared_macros.j2" as shared_macros %}
 {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Remove the following variable (and the condition later in this file) for async rest transport once support for it is GA. #}
 {% set rest_async_io_enabled = api.all_library_settings[api.naming.proto_package].python_settings.experimental_features.rest_async_io_enabled %}
 {% extends '_base.py.j2' %}
@@ -12,9 +13,14 @@ except ImportError as e:  # pragma: NO COVER
 
 from google.auth.aio import credentials as ga_credentials_async  # type: ignore
 
+from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
+from google.api_core import retry_async as retries
 
-from typing import Any, Optional
+
+from typing import Any, Callable, Tuple, Optional, Sequence, Union
+
+{{ shared_macros.operations_mixin_imports(api, service, opts) }}
 
 from .rest_base import _Base{{ service.name }}RestTransport
 
@@ -79,6 +85,22 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             api_audience=None
         )
         self._session = AsyncAuthorizedSession(self._credentials)
+        self._wrap_with_kind = True
+        self._prep_wrapped_messages(client_info)
+
+    {{ shared_macros.prep_wrapped_messages_async_method(service)|indent(4) }}
+
+    {{ shared_macros.wrap_async_method_macro()|indent(4) }}
+
+    {% for method in service.methods.values()|sort(attribute="name") %}
+
+    {# TODO(b/362949446) Return a callable and update return type once implemented. #}
+    @property
+    def {{method.transport_safe_name|snake_case}}(self) -> None:
+
+        return
+
+    {% endfor %}
 
     @property
     def kind(self) -> str:

tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py

@@ -20,9 +20,17 @@
 
 from google.auth.aio import credentials as ga_credentials_async  # type: ignore
 
+from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
+from google.api_core import retry_async as retries
+
+
+from typing import Any, Callable, Tuple, Optional, Sequence, Union
+
+
+from google.cloud.redis_v1.types import cloud_redis
+from google.longrunning import operations_pb2  # type: ignore
 
-from typing import Any, Optional
 
 from .rest_base import _BaseCloudRedisRestTransport
 
@@ -104,6 +112,128 @@ def __init__(self, *,
             api_audience=None
         )
         self._session = AsyncAuthorizedSession(self._credentials)
+        self._wrap_with_kind = True
+        self._prep_wrapped_messages(client_info)
+
+    def _prep_wrapped_messages(self, client_info):
+        """ Precompute the wrapped methods, overriding the base class method to use async wrappers."""
+        self._wrapped_methods = {
+            self.list_instances: self._wrap_method(
+                self.list_instances,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.get_instance: self._wrap_method(
+                self.get_instance,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.get_instance_auth_string: self._wrap_method(
+                self.get_instance_auth_string,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.create_instance: self._wrap_method(
+                self.create_instance,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.update_instance: self._wrap_method(
+                self.update_instance,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.upgrade_instance: self._wrap_method(
+                self.upgrade_instance,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.import_instance: self._wrap_method(
+                self.import_instance,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.export_instance: self._wrap_method(
+                self.export_instance,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.failover_instance: self._wrap_method(
+                self.failover_instance,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.delete_instance: self._wrap_method(
+                self.delete_instance,
+                default_timeout=600.0,
+                client_info=client_info,
+            ),
+            self.reschedule_maintenance: self._wrap_method(
+                self.reschedule_maintenance,
+                default_timeout=None,
+                client_info=client_info,
+            ),
+        }
+
+    def _wrap_method(self, func, *args, **kwargs):
+        if self._wrap_with_kind:  # pragma: NO COVER
+            kwargs["kind"] = self.kind
+        return gapic_v1.method_async.wrap_method(func, *args, **kwargs)
+
+    @property
+    def create_instance(self) -> None:
+
+        return
+
+    @property
+    def delete_instance(self) -> None:
+
+        return
+
+    @property
+    def export_instance(self) -> None:
+
+        return
+
+    @property
+    def failover_instance(self) -> None:
+
+        return
+
+    @property
+    def get_instance(self) -> None:
+
+        return
+
+    @property
+    def get_instance_auth_string(self) -> None:
+
+        return
+
+    @property
+    def import_instance(self) -> None:
+
+        return
+
+    @property
+    def list_instances(self) -> None:
+
+        return
+
+    @property
+    def reschedule_maintenance(self) -> None:
+
+        return
+
+    @property
+    def update_instance(self) -> None:
+
+        return
+
+    @property
+    def upgrade_instance(self) -> None:
+
+        return
 
     @property
     def kind(self) -> str: