Copybara import of the project:

--
60b6b75 by Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>:

feat: add v1beta1 StreamingFetchFeatureValues API

PiperOrigin-RevId: 615281865

Source-Link: googleapis/googleapis@f9f3c04

Source-Link: googleapis/googleapis-gen@1351a69
Copy-Tag: eyJwIjoiLmdpdGh1Yi8uT3dsQm90LnlhbWwiLCJoIjoiMTM1MWE2OTgxYzkwMWMyOWM0NDZiYjNjOWQwODNjMGU3ZGI4NTEwOCJ9

--
7bb47f4 by Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>:

🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

COPYBARA_INTEGRATE_REVIEW=#3428 from googleapis:owl-bot-copy 7bb47f4
PiperOrigin-RevId: 617333658

Author: gcf-owl-bot[bot]

google/cloud/aiplatform_v1beta1/__init__.py

@@ -248,6 +248,8 @@
 from .types.feature_online_store_service import NearestNeighbors
 from .types.feature_online_store_service import SearchNearestEntitiesRequest
 from .types.feature_online_store_service import SearchNearestEntitiesResponse
+from .types.feature_online_store_service import StreamingFetchFeatureValuesRequest
+from .types.feature_online_store_service import StreamingFetchFeatureValuesResponse
 from .types.feature_online_store_service import FeatureViewDataFormat
 from .types.feature_registry_service import CreateFeatureGroupOperationMetadata
 from .types.feature_registry_service import CreateFeatureGroupRequest
@@ -1350,6 +1352,8 @@
     "StreamDirectPredictResponse",
     "StreamDirectRawPredictRequest",
     "StreamDirectRawPredictResponse",
+    "StreamingFetchFeatureValuesRequest",
+    "StreamingFetchFeatureValuesResponse",
     "StreamingPredictRequest",
     "StreamingPredictResponse",
     "StreamingRawPredictRequest",

google/cloud/aiplatform_v1beta1/gapic_metadata.json

@@ -755,6 +755,11 @@
               "methods": [
                 "search_nearest_entities"
               ]
+            },
+            "StreamingFetchFeatureValues": {
+              "methods": [
+                "streaming_fetch_feature_values"
+              ]
             }
           }
         },
@@ -770,6 +775,11 @@
               "methods": [
                 "search_nearest_entities"
               ]
+            },
+            "StreamingFetchFeatureValues": {
+              "methods": [
+                "streaming_fetch_feature_values"
+              ]
             }
           }
         },
@@ -785,6 +795,11 @@
               "methods": [
                 "search_nearest_entities"
               ]
+            },
+            "StreamingFetchFeatureValues": {
+              "methods": [
+                "streaming_fetch_feature_values"
+              ]
             }
           }
         }

google/cloud/aiplatform_v1beta1/services/feature_online_store_service/async_client.py

@@ -22,6 +22,9 @@
     MutableMapping,
     MutableSequence,
     Optional,
+    AsyncIterable,
+    Awaitable,
+    AsyncIterator,
     Sequence,
     Tuple,
     Type,
@@ -48,6 +51,7 @@
 from google.iam.v1 import policy_pb2  # type: ignore
 from google.longrunning import operations_pb2  # type: ignore
 from google.protobuf import struct_pb2  # type: ignore
+from google.rpc import status_pb2  # type: ignore
 from .transports.base import FeatureOnlineStoreServiceTransport, DEFAULT_CLIENT_INFO
 from .transports.grpc_asyncio import FeatureOnlineStoreServiceGrpcAsyncIOTransport
 from .client import FeatureOnlineStoreServiceClient
@@ -391,6 +395,107 @@ async def sample_fetch_feature_values():
         # Done; return the response.
         return response
 
+    def streaming_fetch_feature_values(
+        self,
+        requests: Optional[
+            AsyncIterator[
+                feature_online_store_service.StreamingFetchFeatureValuesRequest
+            ]
+        ] = None,
+        *,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Union[float, object] = gapic_v1.method.DEFAULT,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> Awaitable[
+        AsyncIterable[feature_online_store_service.StreamingFetchFeatureValuesResponse]
+    ]:
+        r"""Bidirectional streaming RPC to fetch feature values
+        under a FeatureView. Requests may not have a one-to-one
+        mapping to responses and responses may be returned
+        out-of-order to reduce latency.
+
+        .. code-block:: python
+
+            # This snippet has been automatically generated and should be regarded as a
+            # code template only.
+            # It will require modifications to work:
+            # - It may require correct/in-range values for request initialization.
+            # - It may require specifying regional endpoints when creating the service
+            #   client as shown in:
+            #   https://googleapis.dev/python/google-api-core/latest/client_options.html
+            from google.cloud import aiplatform_v1beta1
+
+            async def sample_streaming_fetch_feature_values():
+                # Create a client
+                client = aiplatform_v1beta1.FeatureOnlineStoreServiceAsyncClient()
+
+                # Initialize request argument(s)
+                request = aiplatform_v1beta1.StreamingFetchFeatureValuesRequest(
+                    feature_view="feature_view_value",
+                )
+
+                # This method expects an iterator which contains
+                # 'aiplatform_v1beta1.StreamingFetchFeatureValuesRequest' objects
+                # Here we create a generator that yields a single `request` for
+                # demonstrative purposes.
+                requests = [request]
+
+                def request_generator():
+                    for request in requests:
+                        yield request
+
+                # Make the request
+                stream = await client.streaming_fetch_feature_values(requests=request_generator())
+
+                # Handle the response
+                async for response in stream:
+                    print(response)
+
+        Args:
+            requests (AsyncIterator[`google.cloud.aiplatform_v1beta1.types.StreamingFetchFeatureValuesRequest`]):
+                The request object AsyncIterator. Request message for
+                [FeatureOnlineStoreService.StreamingFetchFeatureValues][google.cloud.aiplatform.v1beta1.FeatureOnlineStoreService.StreamingFetchFeatureValues].
+                For the entities requested, all features under the
+                requested feature view will be returned.
+            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:
+            AsyncIterable[google.cloud.aiplatform_v1beta1.types.StreamingFetchFeatureValuesResponse]:
+                Response message for
+                   [FeatureOnlineStoreService.StreamingFetchFeatureValues][google.cloud.aiplatform.v1beta1.FeatureOnlineStoreService.StreamingFetchFeatureValues].
+
+        """
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = gapic_v1.method_async.wrap_method(
+            self._client._transport.streaming_fetch_feature_values,
+            default_timeout=None,
+            client_info=DEFAULT_CLIENT_INFO,
+        )
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata) + (gapic_v1.routing_header.to_grpc_metadata(()),)
+
+        # Validate the universe domain.
+        self._client._validate_universe_domain()
+
+        # Send the request.
+        response = rpc(
+            requests,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        # Done; return the response.
+        return response
+
     async def search_nearest_entities(
         self,
         request: Optional[

google/cloud/aiplatform_v1beta1/services/feature_online_store_service/client.py

@@ -22,6 +22,8 @@
     MutableMapping,
     MutableSequence,
     Optional,
+    Iterable,
+    Iterator,
     Sequence,
     Tuple,
     Type,
@@ -53,6 +55,7 @@
 from google.iam.v1 import policy_pb2  # type: ignore
 from google.longrunning import operations_pb2  # type: ignore
 from google.protobuf import struct_pb2  # type: ignore
+from google.rpc import status_pb2  # type: ignore
 from .transports.base import FeatureOnlineStoreServiceTransport, DEFAULT_CLIENT_INFO
 from .transports.grpc import FeatureOnlineStoreServiceGrpcTransport
 from .transports.grpc_asyncio import FeatureOnlineStoreServiceGrpcAsyncIOTransport
@@ -801,6 +804,101 @@ def sample_fetch_feature_values():
         # Done; return the response.
         return response
 
+    def streaming_fetch_feature_values(
+        self,
+        requests: Optional[
+            Iterator[feature_online_store_service.StreamingFetchFeatureValuesRequest]
+        ] = None,
+        *,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Union[float, object] = gapic_v1.method.DEFAULT,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> Iterable[feature_online_store_service.StreamingFetchFeatureValuesResponse]:
+        r"""Bidirectional streaming RPC to fetch feature values
+        under a FeatureView. Requests may not have a one-to-one
+        mapping to responses and responses may be returned
+        out-of-order to reduce latency.
+
+        .. code-block:: python
+
+            # This snippet has been automatically generated and should be regarded as a
+            # code template only.
+            # It will require modifications to work:
+            # - It may require correct/in-range values for request initialization.
+            # - It may require specifying regional endpoints when creating the service
+            #   client as shown in:
+            #   https://googleapis.dev/python/google-api-core/latest/client_options.html
+            from google.cloud import aiplatform_v1beta1
+
+            def sample_streaming_fetch_feature_values():
+                # Create a client
+                client = aiplatform_v1beta1.FeatureOnlineStoreServiceClient()
+
+                # Initialize request argument(s)
+                request = aiplatform_v1beta1.StreamingFetchFeatureValuesRequest(
+                    feature_view="feature_view_value",
+                )
+
+                # This method expects an iterator which contains
+                # 'aiplatform_v1beta1.StreamingFetchFeatureValuesRequest' objects
+                # Here we create a generator that yields a single `request` for
+                # demonstrative purposes.
+                requests = [request]
+
+                def request_generator():
+                    for request in requests:
+                        yield request
+
+                # Make the request
+                stream = client.streaming_fetch_feature_values(requests=request_generator())
+
+                # Handle the response
+                for response in stream:
+                    print(response)
+
+        Args:
+            requests (Iterator[google.cloud.aiplatform_v1beta1.types.StreamingFetchFeatureValuesRequest]):
+                The request object iterator. Request message for
+                [FeatureOnlineStoreService.StreamingFetchFeatureValues][google.cloud.aiplatform.v1beta1.FeatureOnlineStoreService.StreamingFetchFeatureValues].
+                For the entities requested, all features under the
+                requested feature view will be returned.
+            retry (google.api_core.retry.Retry): 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:
+            Iterable[google.cloud.aiplatform_v1beta1.types.StreamingFetchFeatureValuesResponse]:
+                Response message for
+                   [FeatureOnlineStoreService.StreamingFetchFeatureValues][google.cloud.aiplatform.v1beta1.FeatureOnlineStoreService.StreamingFetchFeatureValues].
+
+        """
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[
+            self._transport.streaming_fetch_feature_values
+        ]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata) + (gapic_v1.routing_header.to_grpc_metadata(()),)
+
+        # Validate the universe domain.
+        self._validate_universe_domain()
+
+        # Send the request.
+        response = rpc(
+            requests,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        # Done; return the response.
+        return response
+
     def search_nearest_entities(
         self,
         request: Optional[

google/cloud/aiplatform_v1beta1/services/feature_online_store_service/transports/base.py

@@ -136,6 +136,11 @@ def _prep_wrapped_messages(self, client_info):
                 default_timeout=None,
                 client_info=client_info,
             ),
+            self.streaming_fetch_feature_values: gapic_v1.method.wrap_method(
+                self.streaming_fetch_feature_values,
+                default_timeout=None,
+                client_info=client_info,
+            ),
             self.search_nearest_entities: gapic_v1.method.wrap_method(
                 self.search_nearest_entities,
                 default_timeout=None,
@@ -164,6 +169,18 @@ def fetch_feature_values(
     ]:
         raise NotImplementedError()
 
+    @property
+    def streaming_fetch_feature_values(
+        self,
+    ) -> Callable[
+        [feature_online_store_service.StreamingFetchFeatureValuesRequest],
+        Union[
+            feature_online_store_service.StreamingFetchFeatureValuesResponse,
+            Awaitable[feature_online_store_service.StreamingFetchFeatureValuesResponse],
+        ],
+    ]:
+        raise NotImplementedError()
+
     @property
     def search_nearest_entities(
         self,

google/cloud/aiplatform_v1beta1/services/feature_online_store_service/transports/grpc.py

@@ -262,6 +262,40 @@ def fetch_feature_values(
             )
         return self._stubs["fetch_feature_values"]
 
+    @property
+    def streaming_fetch_feature_values(
+        self,
+    ) -> Callable[
+        [feature_online_store_service.StreamingFetchFeatureValuesRequest],
+        feature_online_store_service.StreamingFetchFeatureValuesResponse,
+    ]:
+        r"""Return a callable for the streaming fetch feature values method over gRPC.
+
+        Bidirectional streaming RPC to fetch feature values
+        under a FeatureView. Requests may not have a one-to-one
+        mapping to responses and responses may be returned
+        out-of-order to reduce latency.
+
+        Returns:
+            Callable[[~.StreamingFetchFeatureValuesRequest],
+                    ~.StreamingFetchFeatureValuesResponse]:
+                A function that, when called, will call the underlying RPC
+                on the server.
+        """
+        # Generate a "stub function" on-the-fly which will actually make
+        # the request.
+        # gRPC handles serialization and deserialization, so we just need
+        # to pass in the functions for each.
+        if "streaming_fetch_feature_values" not in self._stubs:
+            self._stubs[
+                "streaming_fetch_feature_values"
+            ] = self.grpc_channel.stream_stream(
+                "/google.cloud.aiplatform.v1beta1.FeatureOnlineStoreService/StreamingFetchFeatureValues",
+                request_serializer=feature_online_store_service.StreamingFetchFeatureValuesRequest.serialize,
+                response_deserializer=feature_online_store_service.StreamingFetchFeatureValuesResponse.deserialize,
+            )
+        return self._stubs["streaming_fetch_feature_values"]
+
     @property
     def search_nearest_entities(
         self,