feat(generator): add _compat.py fallback for transcoding rest_helpers

Author: hebaalazzeh

packages/gapic-generator/gapic/generator/generator.py

@@ -120,7 +120,7 @@ def get_response(self, api_schema: api.API, opts: Options) -> CodeGeneratorRespo
         for template_name in client_templates:
             # Quick check: Skip "private" templates.
             filename = template_name.split("/")[-1]
-            if filename.startswith("_") and filename != "__init__.py.j2":
+            if filename.startswith("_") and filename not in ("__init__.py.j2", "_compat.py.j2"):
                 continue
 
             # Append to the output files dictionary.

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2

@@ -0,0 +1,107 @@
+# {% include '_license.j2' %}
+
+from typing import Any, Dict, List, Optional, Tuple
+
+try:
+    from google.api_core import rest_helpers
+    # Trigger fallback if rest_helpers is an older version missing 'transcode'
+    if not hasattr(rest_helpers, "transcode"):
+        raise ImportError
+except ImportError:
+    # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+    import functools
+    import json
+    import operator
+    from google.protobuf import json_format  # type: ignore
+    from google.api_core import path_template  # type: ignore
+
+    class _FallbackRestHelpers:
+        @staticmethod
+        def flatten_query_params(obj, strict=False):
+            if obj is not None and not isinstance(obj, dict):
+                raise TypeError("flatten_query_params must be called with dict object")
+            return _FallbackRestHelpers._flatten(obj, key_path=[], strict=strict)
+
+        @staticmethod
+        def _flatten(obj, key_path, strict=False):
+            if obj is None:
+                return []
+            if isinstance(obj, dict):
+                return _FallbackRestHelpers._flatten_dict(obj, key_path=key_path, strict=strict)
+            if isinstance(obj, list):
+                return _FallbackRestHelpers._flatten_list(obj, key_path=key_path, strict=strict)
+            return _FallbackRestHelpers._flatten_value(obj, key_path=key_path, strict=strict)
+
+        @staticmethod
+        def _is_primitive_value(obj):
+            if obj is None:
+                return False
+            if isinstance(obj, (list, dict)):
+                raise ValueError("query params may not contain repeated dicts or lists")
+            return True
+
+        @staticmethod
+        def _flatten_value(obj, key_path, strict=False):
+            return [(".".join(key_path), _FallbackRestHelpers._canonicalize(obj, strict=strict))]
+
+        @staticmethod
+        def _flatten_dict(obj, key_path, strict=False):
+            items = (
+                _FallbackRestHelpers._flatten(value, key_path=key_path + [key], strict=strict)
+                for key, value in obj.items()
+            )
+            return functools.reduce(operator.concat, items, [])
+
+        @staticmethod
+        def _flatten_list(elems, key_path, strict=False):
+            items = (
+                _FallbackRestHelpers._flatten_value(elem, key_path=key_path, strict=strict)
+                for elem in elems
+                if _FallbackRestHelpers._is_primitive_value(elem)
+            )
+            return functools.reduce(operator.concat, items, [])
+
+        @staticmethod
+        def _canonicalize(obj, strict=False):
+            if strict:
+                value = str(obj)
+                if isinstance(obj, bool):
+                    value = value.lower()
+                return value
+            return obj
+
+        @staticmethod
+        def transcode(
+            http_options: List[Dict[str, str]],
+            request: Any,
+            required_fields_default_values: Optional[Dict[str, Any]] = None,
+            rest_numeric_enums: bool = False,
+        ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+            pb_request = getattr(request, "_pb", request)
+            transcoded_request = path_template.transcode(http_options, pb_request)
+
+            body_json = None
+            if transcoded_request.get("body") is not None:
+                body_json = json_format.MessageToJson(
+                    transcoded_request["body"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+
+            query_params_json = {}
+            if transcoded_request.get("query_params") is not None:
+                query_params_json = json.loads(json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                ))
+
+            if required_fields_default_values:
+                for k, v in required_fields_default_values.items():
+                    if k not in query_params_json:
+                        query_params_json[k] = v
+
+            if rest_numeric_enums:
+                query_params_json["$alt"] = "json;enum-encoding=int"
+
+            return transcoded_request, body_json, query_params_json
+
+    rest_helpers = _FallbackRestHelpers

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2

@@ -10,7 +10,7 @@ from google.auth.transport.requests import AuthorizedSession  # type: ignore
 from google.auth import credentials as ga_credentials  # type: ignore
 from google.api_core import exceptions as core_exceptions
 from google.api_core import retry as retries
-from google.api_core import rest_helpers
+from .. import _compat as rest_helpers
 from google.api_core import rest_streaming
 from google.api_core import gapic_v1
 import google.protobuf

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

@@ -32,7 +32,7 @@ from google.iam.v1 import policy_pb2  # type: ignore
 from google.cloud.location import locations_pb2 # type: ignore
 {% endif %}
 from google.api_core import retry_async as retries
-from google.api_core import rest_helpers
+from .. import _compat as rest_helpers
 from google.api_core import rest_streaming_async  # type: ignore
 import google.protobuf