Skip to content

feat: add async rest transport call methods #2140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ohmayr merged 1 commit into from
Sep 13, 2024
Merged

feat: add async rest transport call methods #2140

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
33 changes: 30 additions & 3 deletions .../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 @@ -6,7 +6,7 @@
{% block content %}

try:
from google.auth.aio.transport.sessions import AsyncAuthorizedSession # type: ignore
from google.auth.aio.transport.sessions import AsyncAuthorizedSession # type: ignore
except ImportError as e: # pragma: NO COVER
{# TODO(https://github.com/googleapis/google-auth-library-python/pull/1577): Update the version of google-auth once the linked PR is merged. #}
raise ImportError("async rest transport requires google.auth >= 2.x.x") from e
Expand All @@ -17,6 +17,7 @@ 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

import dataclasses
from typing import Any, Callable, Tuple, Optional, Sequence, Union

{{ shared_macros.operations_mixin_imports(api, service, opts) }}
Expand All @@ -25,13 +26,24 @@ from .rest_base import _Base{{ service.name }}RestTransport

from .base import DEFAULT_CLIENT_INFO as BASE_DEFAULT_CLIENT_INFO

try:
OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
except AttributeError: # pragma: NO COVER
OptionalRetry = Union[retries.AsyncRetry, object, None] # type: ignore

{# TODO (https://github.com/googleapis/gapic-generator-python/issues/2128): Update `rest_version` to include the transport dependency version. #}
DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
grpc_version=None,
rest_version=None,
)

{# TODO: Add an `_interceptor` property once implemented #}
Copy link
Contributor

@parthea parthea Sep 11, 2024

Choose a reason for hiding this comment

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

Please include the TODO bug to follow up

Copy link
Contributor Author

@ohmayr ohmayr Sep 12, 2024

Choose a reason for hiding this comment

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

I think it's fine to ignore this one since this comment is addressed and removed in a follow up PR: #2151.

@dataclasses.dataclass
class Async{{service.name}}RestStub:
_session: AsyncAuthorizedSession
_host: str

class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
"""Asynchronous REST backend transport for {{ service.name }}.

Expand Down Expand Up @@ -92,14 +104,29 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
{{ shared_macros.wrap_async_method_macro()|indent(4) }}

{% for method in service.methods.values()|sort(attribute="name") %}
class _{{method.name}}(_Base{{ service.name }}RestTransport._Base{{method.name}}, Async{{service.name}}RestStub):
def __hash__(self):
return hash("Async{{service.name}}RestTransport.{{method.name}}")

async def __call__(self,
request: {{method.input.ident}}, *,
Copy link
Contributor

@vchudnov-g vchudnov-g Sep 12, 2024

Choose a reason for hiding this comment

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

What does the asterisk mean here?

Copy link
Contributor Author

@ohmayr ohmayr Sep 12, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

* enforces that arguments following request must be passed in as keyword arguments.

Copy link
Contributor

@vchudnov-g vchudnov-g Sep 13, 2024

Choose a reason for hiding this comment

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

Oh, it's a Python thing. Because it was on the same line, I assumed it was part of the type hint and glossed over the comma.

I suggest having the * in its own line, like the other parameters. It gets lost here.

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:
raise NotImplementedError(
"Method {{ method.name }} is not available over REST transport"
)

{# TODO(b/362949446) Return a callable once the class is implemented. #}
{% endfor %}
{% for method in service.methods.values()|sort(attribute="name") %}
{# TODO(https://github.com/googleapis/gapic-generator-python/issues/2154): Remove `type: ignore`. #}
@property
def {{method.transport_safe_name|snake_case}}(self) -> Callable[
[{{method.input.ident}}],
{{method.output.ident}}]:
return # type: ignore
return self._{{method.name}}(self._session, self._host) # type: ignore
Copy link
Contributor

@parthea parthea Sep 11, 2024

Choose a reason for hiding this comment

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

Do we still need the # type: ignore ?

Copy link
Contributor Author

@ohmayr ohmayr Sep 12, 2024

Choose a reason for hiding this comment

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

Yes. My initial thought was that it's probably a mismatch since I wasn't returning a Callable. However, I later realised that we're doing the same in rest.py.

I haven't really looked into why we've been ignoring the type in rest.py but I'm following whatever we're doing there. Probably because the Callables for client streaming which raise NotImplmentedError do not return anything so mypy isn't happy about it? Regardless, I've filed an issue to investigate and remove this as a follow up.

Copy link
Contributor

@vchudnov-g vchudnov-g Sep 12, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

So each of these properties is a callable that invokes the appropriate inner class. When do the self._wrapped_methods get called? (And which of your PRs will ensure that?)

Copy link
Contributor Author

@ohmayr ohmayr Sep 13, 2024

Choose a reason for hiding this comment

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

wrapping happens during transport initialization / construction via self._prep_wrapped_messages(client_info). If you look at the implementation for this helper, it creates a dictionary called _wrapped_methods which is something like {method : wrapped_method, ... }.

At the client level instead of calling method(), we call self._wrapped_methods[method](). However, we're not doing this for the following methods:

  • set_iam_policy
  • get_iam_policy
  • test_iam_permissions

FYI these are mixin methods. We don't pre-wrap mixin methods which is why this logic isn't added. We probably should do that or at least pass down the kind property to avoid gRPC code path in a rest call at the very least.

Copy link
Contributor Author

@ohmayr ohmayr Sep 13, 2024

Choose a reason for hiding this comment

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

I think this is important. I've filed #2156 to ensure that we do this as a follow up since this PR doesn't touch the client layer.

Copy link
Contributor

@vchudnov-g vchudnov-g Sep 13, 2024

Choose a reason for hiding this comment

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

At the client level instead of calling method(), we call self._wrapped_methods[method]()

OK, this is what I was looking for.

Once we have self._wrapped_methods, do we ever call a method() directly? If we don't, then the most elegant thing would arguably be to overwrite the properties with their wrapped versions, and just called each (now-wrapped) method() directly. Does that make sense? If so, it would be a good code simplification to do later.


{% endfor %}

Expand Down
2 changes: 0 additions & 2 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 @@ -1989,7 +1989,6 @@ def test_unsupported_parameter_rest_asyncio():
# which are not supported for rest transport.
#}
{% macro rest_method_not_implemented_error(service, method, transport, is_async) %}
{% if not is_async %}{# TODO(b/362949446): Remove this guard once a not implemented __call__ class method is added to async rest for every wrapper.Method. #}
{% set await_prefix = get_await_prefix(is_async) %}
{% set async_prefix = get_async_prefix(is_async) %}
{% set async_decorator = get_async_decorator(is_async) %}
Expand All @@ -2015,7 +2014,6 @@ def test_unsupported_parameter_rest_asyncio():
in str(not_implemented_error.value)
)

{% endif %}{# if is_async #}
{% endmacro %}

{# initialize_client_with_transport_test adds coverage for transport clients.
Expand Down
189 changes: 177 additions & 12 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 @@ -14,7 +14,7 @@
# limitations under the License.
#
try:
from google.auth.aio.transport.sessions import AsyncAuthorizedSession # type: ignore
from google.auth.aio.transport.sessions import AsyncAuthorizedSession # type: ignore
except ImportError as e: # pragma: NO COVER
raise ImportError("async rest transport requires google.auth >= 2.x.x") from e

Expand All @@ -24,6 +24,7 @@
from google.api_core import gapic_v1
from google.api_core import retry_async as retries

import dataclasses
from typing import Any, Callable, Tuple, Optional, Sequence, Union


Expand All @@ -35,12 +36,22 @@

from .base import DEFAULT_CLIENT_INFO as BASE_DEFAULT_CLIENT_INFO

try:
OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
except AttributeError: # pragma: NO COVER
OptionalRetry = Union[retries.AsyncRetry, object, None] # type: ignore

DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
grpc_version=None,
rest_version=None,
)

@dataclasses.dataclass
class AsyncCloudRedisRestStub:
_session: AsyncAuthorizedSession
_host: str

class AsyncCloudRedisRestTransport(_BaseCloudRedisRestTransport):
"""Asynchronous REST backend transport for CloudRedis.

Expand Down Expand Up @@ -179,71 +190,225 @@ def _wrap_method(self, func, *args, **kwargs):
kwargs["kind"] = self.kind
return gapic_v1.method_async.wrap_method(func, *args, **kwargs)

class _CreateInstance(_BaseCloudRedisRestTransport._BaseCreateInstance, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.CreateInstance")

async def __call__(self,
request: cloud_redis.CreateInstanceRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method CreateInstance is not available over REST transport"
)

class _DeleteInstance(_BaseCloudRedisRestTransport._BaseDeleteInstance, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.DeleteInstance")

async def __call__(self,
request: cloud_redis.DeleteInstanceRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method DeleteInstance is not available over REST transport"
)

class _ExportInstance(_BaseCloudRedisRestTransport._BaseExportInstance, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.ExportInstance")

async def __call__(self,
request: cloud_redis.ExportInstanceRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method ExportInstance is not available over REST transport"
)

class _FailoverInstance(_BaseCloudRedisRestTransport._BaseFailoverInstance, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.FailoverInstance")

async def __call__(self,
request: cloud_redis.FailoverInstanceRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method FailoverInstance is not available over REST transport"
)

class _GetInstance(_BaseCloudRedisRestTransport._BaseGetInstance, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.GetInstance")

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"
)

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

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"
)

class _ImportInstance(_BaseCloudRedisRestTransport._BaseImportInstance, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.ImportInstance")

async def __call__(self,
request: cloud_redis.ImportInstanceRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method ImportInstance is not available over REST transport"
)

class _ListInstances(_BaseCloudRedisRestTransport._BaseListInstances, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.ListInstances")

async def __call__(self,
request: cloud_redis.ListInstancesRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method ListInstances is not available over REST transport"
)

class _RescheduleMaintenance(_BaseCloudRedisRestTransport._BaseRescheduleMaintenance, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.RescheduleMaintenance")

async def __call__(self,
request: cloud_redis.RescheduleMaintenanceRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method RescheduleMaintenance is not available over REST transport"
)

class _UpdateInstance(_BaseCloudRedisRestTransport._BaseUpdateInstance, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.UpdateInstance")

async def __call__(self,
request: cloud_redis.UpdateInstanceRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method UpdateInstance is not available over REST transport"
)

class _UpgradeInstance(_BaseCloudRedisRestTransport._BaseUpgradeInstance, AsyncCloudRedisRestStub):
def __hash__(self):
return hash("AsyncCloudRedisRestTransport.UpgradeInstance")

async def __call__(self,
request: cloud_redis.UpgradeInstanceRequest, *,
retry: OptionalRetry=gapic_v1.method.DEFAULT,
timeout: Optional[float]=None,
metadata: Sequence[Tuple[str, str]]=(),
) -> None:
raise NotImplementedError(
"Method UpgradeInstance is not available over REST transport"
)

@property
def create_instance(self) -> Callable[
[cloud_redis.CreateInstanceRequest],
operations_pb2.Operation]:
return # type: ignore
return self._CreateInstance(self._session, self._host) # type: ignore

@property
def delete_instance(self) -> Callable[
[cloud_redis.DeleteInstanceRequest],
operations_pb2.Operation]:
return # type: ignore
return self._DeleteInstance(self._session, self._host) # type: ignore

@property
def export_instance(self) -> Callable[
[cloud_redis.ExportInstanceRequest],
operations_pb2.Operation]:
return # type: ignore
return self._ExportInstance(self._session, self._host) # type: ignore

@property
def failover_instance(self) -> Callable[
[cloud_redis.FailoverInstanceRequest],
operations_pb2.Operation]:
return # type: ignore
return self._FailoverInstance(self._session, self._host) # type: ignore

@property
def get_instance(self) -> Callable[
[cloud_redis.GetInstanceRequest],
cloud_redis.Instance]:
return # type: ignore
return self._GetInstance(self._session, self._host) # type: ignore

@property
def get_instance_auth_string(self) -> Callable[
[cloud_redis.GetInstanceAuthStringRequest],
cloud_redis.InstanceAuthString]:
return # type: ignore
return self._GetInstanceAuthString(self._session, self._host) # type: ignore

@property
def import_instance(self) -> Callable[
[cloud_redis.ImportInstanceRequest],
operations_pb2.Operation]:
return # type: ignore
return self._ImportInstance(self._session, self._host) # type: ignore

@property
def list_instances(self) -> Callable[
[cloud_redis.ListInstancesRequest],
cloud_redis.ListInstancesResponse]:
return # type: ignore
return self._ListInstances(self._session, self._host) # type: ignore

@property
def reschedule_maintenance(self) -> Callable[
[cloud_redis.RescheduleMaintenanceRequest],
operations_pb2.Operation]:
return # type: ignore
return self._RescheduleMaintenance(self._session, self._host) # type: ignore

@property
def update_instance(self) -> Callable[
[cloud_redis.UpdateInstanceRequest],
operations_pb2.Operation]:
return # type: ignore
return self._UpdateInstance(self._session, self._host) # type: ignore

@property
def upgrade_instance(self) -> Callable[
[cloud_redis.UpgradeInstanceRequest],
operations_pb2.Operation]:
return # type: ignore
return self._UpgradeInstance(self._session, self._host) # type: ignore

@property
def kind(self) -> str:
Expand Down
Loading