Skip to content
This repository was archived by the owner on Mar 26, 2026. It is now read-only.

feat: implement a basic async rest method #2145

Merged
ohmayr merged 3 commits into from
Sep 17, 2024
Merged

feat: implement a basic async rest method #2145

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9efcb8a
feat: implement a basic async rest method
ohmayr Sep 11, 2024
25d5dd1
address PR comments
ohmayr Sep 17, 2024
4e79325
update auth version in test
ohmayr Sep 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 #}
Expand Down
52 changes: 51 additions & 1 deletion .../templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,11 @@ 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 google.api_core import rest_helpers

from google.protobuf import json_format

import json # type: ignore
import dataclasses
from typing import Any, Callable, Tuple, Optional, Sequence, Union

Expand Down Expand Up @@ -109,16 +113,62 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
def __hash__(self):
return hash("Async{{service.name}}RestTransport.{{method.name}}")

{% if method.http_options and not method.client_streaming and not method.server_streaming and not method.lro and not method.extended_lro and not method.paged_result_field %}
Comment thread
parthea marked this conversation as resolved.
{% set body_spec = method.http_options[0].body %}
{{ shared_macros.response_method(body_spec, is_async=True)|indent(8) }}

{% endif %}{# method.http_options and not method.client_streaming and not method.server_streaming and not method.lro and not method.extended_lro and not method.paged_result_field #}
async def __call__(self,
request: {{method.input.ident}}, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
{# TODO(b/362949446): Update the return type as we implement this for different method types. #}
) -> None:
){% if not method.void %} -> {% if not method.server_streaming %}{{method.output.ident}}{% else %}None{% endif %}{% endif %}:
{% if method.http_options and not method.client_streaming and not method.server_streaming and not method.lro and not method.extended_lro and not method.paged_result_field %}
r"""Call the {{- ' ' -}}
{{ (method.name|snake_case).replace('_',' ')|wrap(
width=70, offset=45, indent=8) }}
{{- ' ' -}} method over HTTP.

Args:
request (~.{{ method.input.ident }}):
The request object.{{ ' ' }}
{{- method.input.meta.doc|rst(width=72, indent=16, nl=False) }}
retry (google.api_core.retry_async.AsyncRetry): Designation of what errors, if any,
should be retried.
timeout (float): The timeout for this request.
metadata (Sequence[Tuple[str, str]]): Strings which should be
sent along with the request as metadata.
{% if not method.void %}

Returns:
~.{{ method.output.ident }}:
{{ method.output.meta.doc|rst(width=72, indent=16) }}
{% endif %}
"""

{{ shared_macros.rest_call_method_common(body_spec, method.name, service.name, is_async=True)|indent(8) }}

{% if not method.void %}
# Return the response
resp = {{method.output.ident}}()
{% if method.output.ident.is_proto_plus_type %}
pb_resp = {{method.output.ident}}.pb(resp)
{% else %}
pb_resp = resp
{% endif %}
content = await response.read()
json_format.Parse(content, pb_resp, ignore_unknown_fields=True)
return resp

{% endif %}{# method.void #}

{% else %}
raise NotImplementedError(
"Method {{ method.name }} is not available over REST transport"
)
{% endif %}{# method.http_options and not method.client_streaming and not method.server_streaming and not method.lro and not method.extended_lro and not method.paged_result_field #}

{% endfor %}
{% for method in service.methods.values()|sort(attribute="name") %}
Expand Down
6 changes: 3 additions & 3 deletions gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1940,7 +1940,7 @@ def test_unsupported_parameter_rest_asyncio():
# There are no plans to add support for client streaming.
#}
{% macro is_rest_unsupported_method(method, is_async) %}
{%- if method.client_streaming or is_async -%}
{%- if method.client_streaming or (is_async and (method.server_streaming or method.lro or method.extended_lro or method.paged_result_field)) -%}
{{'True'}}
{%- else -%}
{{'False'}}
Expand Down Expand Up @@ -2230,10 +2230,10 @@ def test_initialize_client_w_{{transport_name}}():
json_return_value = json_format.MessageToJson(return_value)
{% endif %}{# method.void #}
{% if is_async %}
response_value.read = mock.AsyncMock(return_value=b'{}')
response_value.read = mock.AsyncMock(return_value=json_return_value.encode('UTF-8'))
{% else %}{# is_async #}
response_value.content = json_return_value.encode('UTF-8')
{% endif %}
{% endif %}{# is_async #}
req.return_value = response_value
response = {{ await_prefix }}client.{{ method_name }}(request)

Expand Down
158 changes: 141 additions & 17 deletions ...ation/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,11 @@
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 google.api_core import rest_helpers

from google.protobuf import json_format

import json # type: ignore
import dataclasses
from typing import Any, Callable, Tuple, Optional, Sequence, Union

Expand Down Expand Up @@ -200,7 +204,7 @@ async def __call__(self,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
) -> operations_pb2.Operation:
raise NotImplementedError(
"Method CreateInstance is not available over REST transport"
)
Expand All @@ -214,7 +218,7 @@ async def __call__(self,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
) -> operations_pb2.Operation:
raise NotImplementedError(
"Method DeleteInstance is not available over REST transport"
)
Expand All @@ -228,7 +232,7 @@ async def __call__(self,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
) -> operations_pb2.Operation:
raise NotImplementedError(
"Method ExportInstance is not available over REST transport"
)
Expand All @@ -242,7 +246,7 @@ async def __call__(self,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
) -> operations_pb2.Operation:
raise NotImplementedError(
"Method FailoverInstance is not available over REST transport"
)
Expand All @@ -251,29 +255,149 @@ class _GetInstance(_BaseCloudRedisRestTransport._BaseGetInstance, AsyncCloudRedi
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.GetInstance")

@staticmethod
async def _get_response(
host,
metadata,
query_params,
session,
timeout,
transcoded_request,
body=None):

Comment on lines +259 to +267

@parthea parthea Sep 11, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add type hints and docstring for this method

@ohmayr ohmayr Sep 17, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a protected helper which pre-existed and is not a part of this PR's changed. I see that it's now also leveraged for async rest. Regardless, we can address this as a follow up. Let me know if that sounds good!

uri = transcoded_request['uri']
method = transcoded_request['method']
headers = dict(metadata)
headers['Content-Type'] = 'application/json'
response = await getattr(session, method)(
"{host}{uri}".format(host=host, uri=uri),
timeout=timeout,
headers=headers,
params=rest_helpers.flatten_query_params(query_params, strict=True),
)
return response

async def __call__(self,
request: cloud_redis.GetInstanceRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method GetInstance is not available over REST transport"
)
) -> cloud_redis.Instance:
r"""Call the get instance method over HTTP.

Args:
request (~.cloud_redis.GetInstanceRequest):
The request object. Request for
[GetInstance][google.cloud.redis.v1.CloudRedis.GetInstance].
retry (google.api_core.retry_async.AsyncRetry): Designation of what errors, if any,
should be retried.
timeout (float): The timeout for this request.
metadata (Sequence[Tuple[str, str]]): Strings which should be
sent along with the request as metadata.

Returns:
~.cloud_redis.Instance:
A Memorystore for Redis instance.
"""

http_options = _BaseCloudRedisRestTransport._BaseGetInstance._get_http_options()
transcoded_request = _BaseCloudRedisRestTransport._BaseGetInstance._get_transcoded_request(http_options, request)

# Jsonify the query params
query_params = _BaseCloudRedisRestTransport._BaseGetInstance._get_query_params_json(transcoded_request)

# Send the request
response = await AsyncCloudRedisRestTransport._GetInstance._get_response(self._host, metadata, query_params, self._session, timeout, transcoded_request)

# In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
# subclass.
if response.status_code >= 400:
content = await response.read()
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) # type: ignore

# Return the response
resp = cloud_redis.Instance()
pb_resp = cloud_redis.Instance.pb(resp)
content = await response.read()
json_format.Parse(content, pb_resp, ignore_unknown_fields=True)
return resp

class _GetInstanceAuthString(_BaseCloudRedisRestTransport._BaseGetInstanceAuthString, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.GetInstanceAuthString")

@staticmethod
async def _get_response(
host,
metadata,
query_params,
session,
timeout,
transcoded_request,
body=None):

uri = transcoded_request['uri']
method = transcoded_request['method']
headers = dict(metadata)
headers['Content-Type'] = 'application/json'
response = await getattr(session, method)(
"{host}{uri}".format(host=host, uri=uri),
timeout=timeout,
headers=headers,
params=rest_helpers.flatten_query_params(query_params, strict=True),
)
return response

async def __call__(self,
request: cloud_redis.GetInstanceAuthStringRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method GetInstanceAuthString is not available over REST transport"
)
) -> cloud_redis.InstanceAuthString:
r"""Call the get instance auth string method over HTTP.

Args:
request (~.cloud_redis.GetInstanceAuthStringRequest):
The request object. Request for
[GetInstanceAuthString][google.cloud.redis.v1.CloudRedis.GetInstanceAuthString].
retry (google.api_core.retry_async.AsyncRetry): Designation of what errors, if any,
should be retried.
timeout (float): The timeout for this request.
metadata (Sequence[Tuple[str, str]]): Strings which should be
sent along with the request as metadata.

Returns:
~.cloud_redis.InstanceAuthString:
Instance AUTH string details.
"""

http_options = _BaseCloudRedisRestTransport._BaseGetInstanceAuthString._get_http_options()
transcoded_request = _BaseCloudRedisRestTransport._BaseGetInstanceAuthString._get_transcoded_request(http_options, request)

# Jsonify the query params
query_params = _BaseCloudRedisRestTransport._BaseGetInstanceAuthString._get_query_params_json(transcoded_request)

# Send the request
response = await AsyncCloudRedisRestTransport._GetInstanceAuthString._get_response(self._host, metadata, query_params, self._session, timeout, transcoded_request)

# In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
# subclass.
if response.status_code >= 400:
content = await response.read()
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) # type: ignore

# Return the response
resp = cloud_redis.InstanceAuthString()
pb_resp = cloud_redis.InstanceAuthString.pb(resp)
content = await response.read()
json_format.Parse(content, pb_resp, ignore_unknown_fields=True)
return resp

class _ImportInstance(_BaseCloudRedisRestTransport._BaseImportInstance, AsyncCloudRedisRestStub):
def __hash__(self):
Expand All @@ -284,7 +408,7 @@ async def __call__(self,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
) -> operations_pb2.Operation:
raise NotImplementedError(
"Method ImportInstance is not available over REST transport"
)
Expand All @@ -298,7 +422,7 @@ async def __call__(self,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
) -> cloud_redis.ListInstancesResponse:
raise NotImplementedError(
"Method ListInstances is not available over REST transport"
)
Expand All @@ -312,7 +436,7 @@ async def __call__(self,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
) -> operations_pb2.Operation:
raise NotImplementedError(
"Method RescheduleMaintenance is not available over REST transport"
)
Expand All @@ -326,7 +450,7 @@ async def __call__(self,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
) -> operations_pb2.Operation:
raise NotImplementedError(
"Method UpdateInstance is not available over REST transport"
)
Expand All @@ -340,7 +464,7 @@ async def __call__(self,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
) -> operations_pb2.Operation:
raise NotImplementedError(
"Method UpgradeInstance is not available over REST transport"
)
Expand Down
Loading