chore: update type hints for credentials

.github/workflows/tests.yaml

@@ -55,8 +55,9 @@ jobs:
     strategy:
       # Run showcase tests on the lowest and highest supported runtimes
       matrix:
+        # TODO(https://github.com/googleapis/gapic-generator-python/issues/2121) Remove `showcase_w_rest_async` target when async rest is GA.
         python: ["3.7", "3.12"]
-        target: [showcase, showcase_alternative_templates]
+        target: [showcase, showcase_alternative_templates, showcase_w_rest_async]
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4

gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/_shared_macros.j2

@@ -57,6 +57,7 @@ The `try/except` below can be removed once the minimum version of
 try:
     from google.api_core import version_header
     HAS_GOOGLE_API_CORE_VERSION_HEADER = True  # pragma: NO COVER
+{# NOTE: `pragma: NO COVER` is needed since the coverage for presubmits isn't combined. #}
 except ImportError:  # pragma: NO COVER
     HAS_GOOGLE_API_CORE_VERSION_HEADER = False
 {% endif %}{# service_version #}

gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2

@@ -181,7 +181,8 @@ def _get_http_options():
         payload = json.loads(content.decode('utf-8'))
         request_url = "{host}{uri}".format(host=self._host, uri=transcoded_request['uri'])
         method = transcoded_request['method']
-        raise core_exceptions.format_http_response_error(response, method, request_url, payload)
+        {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2137): Remove `type:  ignore` once version check is added for google-api-core. #}
+        raise core_exceptions.format_http_response_error(response, method, request_url, payload)  # type: ignore
         {% else %}
         raise core_exceptions.from_http_response(response)
         {% endif %}{# is_async #}

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

@@ -1,12 +1,14 @@
 {% extends "_base.py.j2" %}
+{# 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 %}
 
 {% block content %}
 {% import "%namespace/%name_%version/%sub/services/%service/_client_macros.j2" as macros %}
 {% import "%namespace/%name_%version/%sub/services/%service/_shared_macros.j2" as shared_macros %}
 
 from collections import OrderedDict
 import re
-from typing import Dict, Callable, Mapping, MutableMapping, MutableSequence, Optional, {% if service.any_server_streaming %}AsyncIterable, Awaitable, {% endif %}{% if service.any_client_streaming %}AsyncIterator, {% endif %}Sequence, Tuple, Type, Union
+from typing import Callable, Dict, Generic, Mapping, MutableMapping, MutableSequence, Optional, {% if service.any_server_streaming %}AsyncIterable, Awaitable, {% endif %}{% if service.any_client_streaming %}AsyncIterator, {% endif %}Sequence, Tuple, Type, TypeVar, Union
 {% if api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|list %}
 import uuid
 {% endif %}
@@ -30,6 +32,12 @@ try:
 except AttributeError:  # pragma: NO COVER
     OptionalRetry = Union[retries.AsyncRetry, object, None]  # type: ignore
 
+try:
+    from google.auth.aio import credentials as ga_credentials_async  # type: ignore
+    CredentialsType = Union[ga_credentials.Credentials, ga_credentials_async.Credentials]  # pragma: NO COVER
+except ImportError:  # pragma: NO COVER
+    CredentialsType = Union[ga_credentials.Credentials, object, None]  # type: ignore
+
 {% filter sort_lines %}
 {% for method in service.methods.values() %}
 {% for ref_type in method.flat_ref_types %}
@@ -174,15 +182,15 @@ class {{ service.async_client_name }}:
     get_transport_class = {{ service.client_name }}.get_transport_class
 
     def __init__(self, *,
-            credentials: Optional[ga_credentials.Credentials] = None,
+            credentials: Optional[CredentialsType] = None,
             transport: Optional[Union[str, {{ service.name }}Transport, Callable[..., {{ service.name }}Transport]]] = "grpc_asyncio",
             client_options: Optional[ClientOptions] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             ) -> None:
         """Instantiates the {{ (service.async_client_name|snake_case).replace("_", " ") }}.
 
         Args:
-            credentials (Optional[google.auth.credentials.Credentials]): The
+            credentials (Optional[google.auth.credentials.Credentials, google.auth.aio.credentials.Credentials]): The
                 authorization credentials to attach to requests. These
                 credentials identify the application to the service; if none
                 are specified, the client will attempt to ascertain the

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

@@ -66,7 +66,14 @@ from .transports.grpc_asyncio import {{ service.grpc_asyncio_transport_name }}
 from .transports.rest import {{ service.name }}RestTransport
 {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Remove this condition when async rest is GA. #}
 {% if rest_async_io_enabled %}
-from .transports.rest_asyncio import Async{{ service.name }}RestTransport
+try:
+    from .transports.rest_asyncio import Async{{ service.name }}RestTransport
+    HAS_ASYNC_REST_DEPENDENCIES = True
+{# NOTE: `pragma: NO COVER` is needed since the coverage for presubmits isn't combined. #}
+except ImportError as e: # pragma: NO COVER
+    HAS_ASYNC_REST_DEPENDENCIES = False
+    ASYNC_REST_EXCEPTION = e
+
 {% endif %}{# if rest_async_io_enabled #}
 {% endif %}
 
@@ -87,7 +94,8 @@ class {{ service.client_name }}Meta(type):
     _transport_registry["rest"] = {{ service.name }}RestTransport
     {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Remove this condition when async rest is GA. #}
     {% if rest_async_io_enabled %}
-    _transport_registry["rest_asyncio"] = Async{{ service.name }}RestTransport
+    if HAS_ASYNC_REST_DEPENDENCIES:  # pragma: NO COVER
+        _transport_registry["rest_asyncio"] = Async{{ service.name }}RestTransport
     {% endif %}{# if rest_async_io_enabled #}
     {% endif %}
 
@@ -104,6 +112,12 @@ class {{ service.client_name }}Meta(type):
             The transport class to use.
         """
         # If a specific transport is requested, return that one.
+        {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Remove this condition when async rest is GA. #}
+        {% if rest_async_io_enabled %}
+        {# NOTE: `pragma: NO COVER` is needed since the coverage for presubmits isn't combined. #}
+        if label == "rest_asyncio" and not HAS_ASYNC_REST_DEPENDENCIES:  # pragma: NO COVER
+            raise ASYNC_REST_EXCEPTION
+        {% endif %}
         if label:
             return cls._transport_registry[label]
 
@@ -552,16 +566,51 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 self._use_mtls_endpoint))
 
         if not transport_provided:
+            {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Remove this condition when async rest is GA. #}
+            {% if rest_async_io_enabled %}
+            transport_init: Union[Type[{{ service.name  }}Transport], Callable[..., {{ service.name }}Transport]] = (
+                {{ service.client_name  }}.get_transport_class(transport)
+                if isinstance(transport, str) or transport is None
+                else cast(Callable[..., {{ service.name }}Transport], transport)
+            )
+
+            if "rest_asyncio" in str(transport_init):
+                {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2136): Support the following parameters in async rest: #}
+                unsupported_params = {
+                    "google.api_core.client_options.ClientOptions.credentials_file": self._client_options.credentials_file,
+                    "google.api_core.client_options.ClientOptions.scopes": self._client_options.scopes,
+                    "google.api_core.client_options.ClientOptions.quota_project_id": self._client_options.quota_project_id,
+                    "google.api_core.client_options.ClientOptions.client_cert_source": self._client_options.client_cert_source,
+                    "google.api_core.client_options.ClientOptions.api_audience": self._client_options.api_audience,
+
+                }
+                provided_unsupported_params = [name for name, value in unsupported_params.items() if value is not None]
+                if provided_unsupported_params:
+                    {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2137): Remove `type: ignore` once we add a version check for google-api-core. #}
+                    raise core_exceptions.AsyncRestUnsupportedParameterError(  # type: ignore
+                        f"The following provided parameters are not supported for `transport=rest_asyncio`: {', '.join(provided_unsupported_params)}"
+                    )
+                self._transport = transport_init(
+                    credentials=credentials,
+                    host=self._api_endpoint,
+                    client_info=client_info,
+                )
+                return
+
+            {% endif %}
             import google.auth._default  # type: ignore
 
             if api_key_value and hasattr(google.auth._default, "get_api_key_credentials"):
                 credentials = google.auth._default.get_api_key_credentials(api_key_value)
 
+            {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Remove this condition when async rest is GA. #}
+            {% if not rest_async_io_enabled %}
             transport_init: Union[Type[{{ service.name  }}Transport], Callable[..., {{ service.name }}Transport]] = (
                 {{ service.client_name  }}.get_transport_class(transport)
                 if isinstance(transport, str) or transport is None
                 else cast(Callable[..., {{ service.name }}Transport], transport)
             )
+            {% endif %}
             # initialize with the provided callable or the passed in class
             self._transport = transport_init(
                 credentials=credentials,

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

@@ -1,9 +1,11 @@
+{# 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' %}
 
 {% block content %}
 
 from collections import OrderedDict
-from typing import Dict, Type
+from typing import Dict, Type{% if rest_async_io_enabled %}, Tuple{% endif +%}
 
 from .base import {{ service.name }}Transport
 {% if 'grpc' in opts.transport %}
@@ -13,6 +15,16 @@ from .grpc_asyncio import {{ service.name }}GrpcAsyncIOTransport
 {% if 'rest' in opts.transport %}
 from .rest import {{ service.name }}RestTransport
 from .rest import {{ service.name }}RestInterceptor
+{% if rest_async_io_enabled %}
+ASYNC_REST_CLASSES: Tuple[str, ...]
+try:
+    from .rest_asyncio import Async{{ service.name }}RestTransport
+    ASYNC_REST_CLASSES = ('Async{{ service.name }}RestTransport',)
+    HAS_REST_ASYNC = True
+except ImportError:  # pragma: NO COVER
+    ASYNC_REST_CLASSES = ()
+    HAS_REST_ASYNC = False
+{% endif %}{# if rest_async_io_enabled #}
 {% endif %}
 
 
@@ -25,6 +37,10 @@ _transport_registry['grpc_asyncio'] = {{ service.name }}GrpcAsyncIOTransport
 {% endif %}
 {% if 'rest' in opts.transport %}
 _transport_registry['rest'] = {{ service.name }}RestTransport
+{% if rest_async_io_enabled %}
+if HAS_REST_ASYNC:  # pragma: NO COVER
+    _transport_registry['rest_asyncio'] = Async{{ service.name }}RestTransport
+{% endif %}{# if rest_async_io_enabled #}
 {% endif %}
 
 __all__ = (
@@ -37,5 +53,5 @@ __all__ = (
     '{{ service.name }}RestTransport',
     '{{ service.name }}RestInterceptor',
     {% endif %}
-)
+){% if 'rest' in opts.transport and rest_async_io_enabled%} + ASYNC_REST_CLASSES{%endif%}
 {% endblock %}