feat: implement OperationsRestAsyncTransport to support long running operations

google/api_core/client_info.py

@@ -57,7 +57,8 @@ class ClientInfo(object):
         user_agent (Optional[str]): Prefix to the user agent header. This is
             used to supply information such as application name or partner tool.
             Recommended format: ``application-or-tool-ID/major.minor.version``.
-        rest_version (Optional[str]): The requests library version.
+        rest_version (Optional[str]): A string with labeled versions of the
+            dependencies used for REST transport.
     """
 
     def __init__(

google/api_core/gapic_v1/client_info.py

@@ -45,6 +45,8 @@ class ClientInfo(client_info.ClientInfo):
         user_agent (Optional[str]): Prefix to the user agent header. This is
             used to supply information such as application name or partner tool.
             Recommended format: ``application-or-tool-ID/major.minor.version``.
+        rest_version (Optional[str]): A string with labeled versions of the
+            dependencies used for REST transport.
     """
 
     def to_grpc_metadata(self):

google/api_core/operations_v1/__init__.py

@@ -25,5 +25,14 @@
     "AbstractOperationsClient",
     "OperationsAsyncClient",
     "OperationsClient",
-    "OperationsRestTransport",
+    "OperationsRestTransport"
 ]
+
+try:
+    from google.api_core.operations_v1.transports.rest_asyncio import OperationsRestAsyncTransport  # noqa: F401
+    __all__.append("OperationsRestAsyncTransport")
+except ImportError:
+    # This import requires the `async_rest` extra.
+    # Don't raise an exception if `OperationsRestAsyncTransport` cannot be imported
+    # as other transports are still available.
+    pass

google/api_core/operations_v1/transports/__init__.py

@@ -14,16 +14,26 @@
 # limitations under the License.
 #
 from collections import OrderedDict
+from typing import cast, Dict, Tuple
 
 from .base import OperationsTransport
 from .rest import OperationsRestTransport
 
-
 # Compile a registry of transports.
-_transport_registry = OrderedDict()
-_transport_registry["rest"] = OperationsRestTransport
+_transport_registry: Dict[str, OperationsTransport] = OrderedDict()
+_transport_registry["rest"] = cast(OperationsTransport, OperationsRestTransport)
+
+__all__: Tuple[str, ...] = ("OperationsTransport", "OperationsRestTransport")
+
+try:
+    from .rest_asyncio import OperationsRestAsyncTransport
 
-__all__ = (
-    "OperationsTransport",
-    "OperationsRestTransport",
-)
+    __all__ += ("OperationsRestAsyncTransport",)
+    _transport_registry["rest_asyncio"] = cast(
+        OperationsTransport, OperationsRestAsyncTransport
+    )
+except ImportError:
+    # This import requires the `async_rest` extra.
+    # Don't raise an exception if `OperationsRestAsyncTransport` cannot be imported
+    # as other transports are still available.
+    pass

google/api_core/operations_v1/transports/base.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 #
 import abc
+import re
 from typing import Awaitable, Callable, Optional, Sequence, Union
 
 import google.api_core  # type: ignore
@@ -25,10 +26,13 @@
 from google.auth import credentials as ga_credentials  # type: ignore
 from google.longrunning import operations_pb2
 from google.oauth2 import service_account  # type: ignore
-from google.protobuf import empty_pb2  # type: ignore
+import google.protobuf
+from google.protobuf import empty_pb2, json_format  # type: ignore
 from grpc import Compression
 
 
+PROTOBUF_VERSION = google.protobuf.__version__
+
 DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
     gapic_version=version.__version__,
 )
@@ -45,12 +49,14 @@ def __init__(
         self,
         *,
         host: str = DEFAULT_HOST,
+        # TODO(https://github.com/googleapis/python-api-core/issues/709): update type hint for credentials to include `google.auth.aio.Credentials`.
         credentials: Optional[ga_credentials.Credentials] = None,
         credentials_file: Optional[str] = None,
         scopes: Optional[Sequence[str]] = None,
         quota_project_id: Optional[str] = None,
         client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
         always_use_jwt_access: Optional[bool] = False,
+        url_scheme="https",
         **kwargs,
     ) -> None:
         """Instantiate the transport.
@@ -76,10 +82,23 @@ def __init__(
                 your own client library.
             always_use_jwt_access (Optional[bool]): Whether self signed JWT should
                 be used for service account credentials.
+            url_scheme: the protocol scheme for the API endpoint.  Normally
+                "https", but for testing or local servers,
+                "http" can be specified.
         """
+        maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host)
+        if maybe_url_match is None:
+            raise ValueError(
+                f"Unexpected hostname structure: {host}"
+            )  # pragma: NO COVER
+
+        url_match_items = maybe_url_match.groupdict()
+
+        host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
+
         # Save the hostname. Default to port 443 (HTTPS) if none is specified.
         if ":" not in host:
-            host += ":443"
+            host += ":443"  # pragma: NO COVER
         self._host = host
 
         scopes_kwargs = {"scopes": scopes, "default_scopes": self.AUTH_SCOPES}
@@ -189,6 +208,37 @@ def close(self):
         """
         raise NotImplementedError()
 
+    def _convert_protobuf_message_to_dict(
+        self, message: google.protobuf.message.Message
+    ):
+        r"""Converts protobuf message to a dictionary.
+
+        When the dictionary is encoded to JSON, it conforms to proto3 JSON spec.
+
+        Args:
+            message(google.protobuf.message.Message): The protocol buffers message
+                instance to serialize.
+
+        Returns:
+            A dict representation of the protocol buffer message.
+        """
+        # TODO(https://github.com/googleapis/python-api-core/issues/643): For backwards compatibility
+        # with protobuf 3.x 4.x, Remove once support for protobuf 3.x and 4.x is dropped.
+        if PROTOBUF_VERSION[0:2] in ["3.", "4."]:
+            result = json_format.MessageToDict(
+                message,
+                preserving_proto_field_name=True,
+                including_default_value_fields=True,  # type: ignore # backward compatibility
+            )
+        else:
+            result = json_format.MessageToDict(
+                message,
+                preserving_proto_field_name=True,
+                always_print_fields_with_no_presence=True,
+            )
+
+        return result
+
     @property
     def list_operations(
         self,

google/api_core/operations_v1/transports/rest.py

@@ -14,7 +14,6 @@
 # limitations under the License.
 #
 
-import re
 from typing import Callable, Dict, Optional, Sequence, Tuple, Union
 
 from requests import __version__ as requests_version
@@ -41,7 +40,7 @@
 DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
     gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
     grpc_version=None,
-    rest_version=requests_version,
+    rest_version=f"requests@{requests_version}",
 )
 
 
@@ -123,16 +122,6 @@ def __init__(
         # TODO(yon-mg): resolve other ctor params i.e. scopes, quota, etc.
         # TODO: When custom host (api_endpoint) is set, `scopes` must *also* be set on the
         # credentials object
-        maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host)
-        if maybe_url_match is None:
-            raise ValueError(
-                f"Unexpected hostname structure: {host}"
-            )  # pragma: NO COVER
-
-        url_match_items = maybe_url_match.groupdict()
-
-        host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
-
         super().__init__(
             host=host,
             credentials=credentials,
@@ -144,6 +133,7 @@ def __init__(
         )
         if client_cert_source_for_mtls:
             self._session.configure_mtls_channel(client_cert_source_for_mtls)
+        # TODO(https://github.com/googleapis/python-api-core/issues/720): Add wrap logic directly to the property methods for callables.
         self._prep_wrapped_messages(client_info)
         self._http_options = http_options or {}
         self._path_prefix = path_prefix
@@ -206,6 +196,7 @@ def _list_operations(
         # Send the request
         headers = dict(metadata)
         headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
         response = getattr(self._session, method)(
             "{host}{uri}".format(host=self._host, uri=uri),
             timeout=timeout,
@@ -282,6 +273,7 @@ def _get_operation(
         # Send the request
         headers = dict(metadata)
         headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
         response = getattr(self._session, method)(
             "{host}{uri}".format(host=self._host, uri=uri),
             timeout=timeout,
@@ -351,6 +343,7 @@ def _delete_operation(
         # Send the request
         headers = dict(metadata)
         headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
         response = getattr(self._session, method)(
             "{host}{uri}".format(host=self._host, uri=uri),
             timeout=timeout,
@@ -426,6 +419,7 @@ def _cancel_operation(
         # Send the request
         headers = dict(metadata)
         headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
         response = getattr(self._session, method)(
             "{host}{uri}".format(host=self._host, uri=uri),
             timeout=timeout,
@@ -441,38 +435,6 @@ def _cancel_operation(
 
         return empty_pb2.Empty()
 
-    def _convert_protobuf_message_to_dict(
-        self, message: google.protobuf.message.Message
-    ):
-        r"""Converts protobuf message to a dictionary.
-
-        When the dictionary is encoded to JSON, it conforms to proto3 JSON spec.
-
-        Args:
-            message(google.protobuf.message.Message): The protocol buffers message
-                instance to serialize.
-
-        Returns:
-            A dict representation of the protocol buffer message.
-        """
-        # For backwards compatibility with protobuf 3.x 4.x
-        # Remove once support for protobuf 3.x and 4.x is dropped
-        # https://github.com/googleapis/python-api-core/issues/643
-        if PROTOBUF_VERSION[0:2] in ["3.", "4."]:
-            result = json_format.MessageToDict(
-                message,
-                preserving_proto_field_name=True,
-                including_default_value_fields=True,  # type: ignore # backward compatibility
-            )
-        else:
-            result = json_format.MessageToDict(
-                message,
-                preserving_proto_field_name=True,
-                always_print_fields_with_no_presence=True,
-            )
-
-        return result
-
     @property
     def list_operations(
         self,