feat: support authorized views (#1034)

Author: daniel-sanche

.cross_sync/transformers.py

@@ -81,7 +81,11 @@ def visit_FunctionDef(self, node):
 
  def visit_Constant(self, node):
  """Replace string type annotations"""
- node.s = self.replacements.get(node.s, node.s)
+ try:
+ node.s = self.replacements.get(node.s, node.s)
+ except TypeError:
+ # ignore unhashable types (e.g. list)
+ pass
  return node
 
 

docs/data_client/async_data_authorized_view.rst

@@ -0,0 +1,11 @@
+Authorized View Async
+~~~~~~~~~~~~~~~~~~~~~
+
+ .. note::
+
+ It is generally not recommended to use the async client in an otherwise synchronous codebase. To make use of asyncio's
+ performance benefits, the codebase should be designed to be async from the ground up.
+
+.. autoclass:: google.cloud.bigtable.data._async.client.AuthorizedViewAsync
+ :members:
+ :inherited-members:

docs/data_client/async_data_table.rst

@@ -8,4 +8,4 @@ Table Async
 
 .. autoclass:: google.cloud.bigtable.data._async.client.TableAsync
  :members:
- :show-inheritance:
+ :inherited-members:

docs/data_client/data_client_usage.rst

@@ -9,6 +9,7 @@ Sync Surface
 
  sync_data_client
  sync_data_table
+ sync_data_authorized_view
  sync_data_mutations_batcher
  sync_data_execute_query_iterator
 
@@ -20,6 +21,7 @@ Async Surface
 
  async_data_client
  async_data_table
+ async_data_authorized_view
  async_data_mutations_batcher
  async_data_execute_query_iterator
 

docs/data_client/sync_data_authorized_view.rst

@@ -0,0 +1,6 @@
+Authorized View
+~~~~~~~~~~~~~~~
+
+.. autoclass:: google.cloud.bigtable.data._sync_autogen.client.AuthorizedView
+ :members:
+ :inherited-members:

google/cloud/bigtable/data/__init__.py

@@ -17,9 +17,11 @@
 
 from google.cloud.bigtable.data._async.client import BigtableDataClientAsync
 from google.cloud.bigtable.data._async.client import TableAsync
+from google.cloud.bigtable.data._async.client import AuthorizedViewAsync
 from google.cloud.bigtable.data._async.mutations_batcher import MutationsBatcherAsync
 from google.cloud.bigtable.data._sync_autogen.client import BigtableDataClient
 from google.cloud.bigtable.data._sync_autogen.client import Table
+from google.cloud.bigtable.data._sync_autogen.client import AuthorizedView
 from google.cloud.bigtable.data._sync_autogen.mutations_batcher import MutationsBatcher
 
 from google.cloud.bigtable.data.read_rows_query import ReadRowsQuery
@@ -76,9 +78,11 @@
 __all__ = (
  "BigtableDataClientAsync",
  "TableAsync",
+ "AuthorizedViewAsync",
  "MutationsBatcherAsync",
  "BigtableDataClient",
  "Table",
+ "AuthorizedView",
  "MutationsBatcher",
  "RowKeySamples",
  "ReadRowsQuery",

google/cloud/bigtable/data/_async/_mutate_rows.py

@@ -15,10 +15,10 @@
 from __future__ import annotations
 
 from typing import Sequence, TYPE_CHECKING
-import functools
 
 from google.api_core import exceptions as core_exceptions
 from google.api_core import retry as retries
+import google.cloud.bigtable_v2.types.bigtable as types_pb
 import google.cloud.bigtable.data.exceptions as bt_exceptions
 from google.cloud.bigtable.data._helpers import _attempt_timeout_generator
 from google.cloud.bigtable.data._helpers import _retry_exception_factory
@@ -36,12 +36,16 @@
  from google.cloud.bigtable_v2.services.bigtable.async_client import (
  BigtableAsyncClient as GapicClientType,
  )
- from google.cloud.bigtable.data._async.client import TableAsync as TableType
+ from google.cloud.bigtable.data._async.client import ( # type: ignore
+ _DataApiTargetAsync as TargetType,
+ )
  else:
  from google.cloud.bigtable_v2.services.bigtable.client import ( # type: ignore
  BigtableClient as GapicClientType,
  )
- from google.cloud.bigtable.data._sync_autogen.client import Table as TableType # type: ignore
+ from google.cloud.bigtable.data._sync_autogen.client import ( # type: ignore
+ _DataApiTarget as TargetType,
+ )
 
 __CROSS_SYNC_OUTPUT__ = "google.cloud.bigtable.data._sync_autogen._mutate_rows"
 
@@ -59,7 +63,7 @@ class _MutateRowsOperationAsync:
 
  Args:
  gapic_client: the client to use for the mutate_rows call
- table: the table associated with the request
+ target: the table or view associated with the request
  mutation_entries: a list of RowMutationEntry objects to send to the server
  operation_timeout: the timeout to use for the entire operation, in seconds.
  attempt_timeout: the timeout to use for each mutate_rows attempt, in seconds.
@@ -70,7 +74,7 @@ class _MutateRowsOperationAsync:
  def __init__(
  self,
  gapic_client: GapicClientType,
- table: TableType,
+ target: TargetType,
  mutation_entries: list["RowMutationEntry"],
  operation_timeout: float,
  attempt_timeout: float | None,
@@ -84,13 +88,8 @@ def __init__(
  f"{_MUTATE_ROWS_REQUEST_MUTATION_LIMIT} mutations across "
  f"all entries. Found {total_mutations}."
  )
- # create partial function to pass to trigger rpc call
- self._gapic_fn = functools.partial(
- gapic_client.mutate_rows,
- table_name=table.table_name,
- app_profile_id=table.app_profile_id,
- retry=None,
- )
+ self._target = target
+ self._gapic_fn = gapic_client.mutate_rows
  # create predicate for determining which errors are retryable
  self.is_retryable = retries.if_exception_type(
  # RPC level errors
@@ -173,8 +172,12 @@ async def _run_attempt(self):
  # make gapic request
  try:
  result_generator = await self._gapic_fn(
+ request=types_pb.MutateRowsRequest(
+ entries=request_entries,
+ app_profile_id=self._target.app_profile_id,
+ **self._target._request_path,
+ ),
  timeout=next(self.timeout_generator),
- entries=request_entries,
  retry=None,
  )
  async for result_list in result_generator:

google/cloud/bigtable/data/_async/_read_rows.py

@@ -37,9 +37,11 @@
 
 if TYPE_CHECKING:
  if CrossSync.is_async:
- from google.cloud.bigtable.data._async.client import TableAsync as TableType
+ from google.cloud.bigtable.data._async.client import (
+ _DataApiTargetAsync as TargetType,
+ )
  else:
- from google.cloud.bigtable.data._sync_autogen.client import Table as TableType # type: ignore
+ from google.cloud.bigtable.data._sync_autogen.client import _DataApiTarget as TargetType # type: ignore
 
 __CROSS_SYNC_OUTPUT__ = "google.cloud.bigtable.data._sync_autogen._read_rows"
 
@@ -59,7 +61,7 @@ class _ReadRowsOperationAsync:
 
  Args:
  query: The query to execute
- table: The table to send the request to
+ target: The table or view to send the request to
  operation_timeout: The total time to allow for the operation, in seconds
  attempt_timeout: The time to allow for each individual attempt, in seconds
  retryable_exceptions: A list of exceptions that should trigger a retry
@@ -69,7 +71,7 @@ class _ReadRowsOperationAsync:
  "attempt_timeout_gen",
  "operation_timeout",
  "request",
- "table",
+ "target",
  "_predicate",
  "_last_yielded_row_key",
  "_remaining_count",
@@ -78,7 +80,7 @@ class _ReadRowsOperationAsync:
  def __init__(
  self,
  query: ReadRowsQuery,
- table: TableType,
+ target: TargetType,
  operation_timeout: float,
  attempt_timeout: float,
  retryable_exceptions: Sequence[type[Exception]] = (),
@@ -90,12 +92,12 @@ def __init__(
  if isinstance(query, dict):
  self.request = ReadRowsRequestPB(
  **query,
- table_name=table.table_name,
- app_profile_id=table.app_profile_id,
+ **target._request_path,
+ app_profile_id=target.app_profile_id,
  )
  else:
- self.request = query._to_pb(table)
- self.table = table
+ self.request = query._to_pb(target)
+ self.target = target
  self._predicate = retries.if_exception_type(*retryable_exceptions)
  self._last_yielded_row_key: bytes | None = None
  self._remaining_count: int | None = self.request.rows_limit or None
@@ -142,7 +144,7 @@ def _read_rows_attempt(self) -> CrossSync.Iterable[Row]:
  if self._remaining_count == 0:
  return self.merge_rows(None)
  # create and return a new row merger
- gapic_stream = self.table.client._gapic_client.read_rows(
+ gapic_stream = self.target.client._gapic_client.read_rows(
  self.request,
  timeout=next(self.attempt_timeout_gen),
  retry=None,