feat: row_exists and read_row

google/cloud/bigtable/client.py

@@ -40,6 +40,7 @@
 )
 from google.cloud.client import ClientWithProject
 from google.api_core.exceptions import GoogleAPICallError
+from google.cloud.bigtable.exceptions import RowNotFound
 from google.api_core import retry_async as retries
 from google.api_core import exceptions as core_exceptions
 from google.cloud.bigtable._read_rows import _ReadRowsOperation
@@ -55,6 +56,10 @@
 from google.cloud.bigtable._helpers import _make_metadata
 from google.cloud.bigtable._helpers import _convert_retry_deadline
 
+from google.cloud.bigtable.row_filters import StripValueTransformerFilter
+from google.cloud.bigtable.row_filters import CellsRowLimitFilter
+from google.cloud.bigtable.row_filters import RowFilterChain
+
 if TYPE_CHECKING:
     from google.cloud.bigtable.mutations_batcher import MutationsBatcher
     from google.cloud.bigtable import RowKeySamples
@@ -508,10 +513,22 @@ async def read_row(
 
         See read_rows_stream
 
+        Raises:
+            - google.cloud.bigtable.exceptions.RowNotFound: if the row does not exist
         Returns:
             - the individual row requested
         """
-        raise NotImplementedError
+        if row_key is None:
+            raise ValueError("row_key must be string or bytes")
+        query = ReadRowsQuery(row_keys=row_key, limit=1)
+        results = await self.read_rows(
+            query,
+            operation_timeout=operation_timeout,
+            per_request_timeout=per_request_timeout,
+        )
+        if len(results) == 0:
+            raise RowNotFound(f"Row {row_key!r} not found")
+        return results[0]
 
     async def read_rows_sharded(
         self,
@@ -547,7 +564,18 @@ async def row_exists(
         Returns:
             - a bool indicating whether the row exists
         """
-        raise NotImplementedError
+        if row_key is None:
+            raise ValueError("row_key must be string or bytes")
+        strip_filter = StripValueTransformerFilter(flag=True)
+        limit_filter = CellsRowLimitFilter(1)
+        chain_filter = RowFilterChain(filters=[limit_filter, strip_filter])
+        query = ReadRowsQuery(row_keys=row_key, limit=1, row_filter=chain_filter)
+        results = await self.read_rows(
+            query,
+            operation_timeout=operation_timeout,
+            per_request_timeout=per_request_timeout,
+        )
+        return len(results) > 0
 
     async def sample_keys(
         self,

google/cloud/bigtable/exceptions.py

@@ -39,6 +39,10 @@ class InvalidChunk(core_exceptions.GoogleAPICallError):
     """Exception raised to invalid chunk data from back-end."""
 
 
+class RowNotFound(core_exceptions.NotFound):
+    """Exception raised when a row is not found on a read_row call."""
+
+
 class _RowSetComplete(Exception):
     """
     Internal exception for _ReadRowsOperation

google/cloud/bigtable/read_rows_query.py

@@ -106,12 +106,12 @@ def __init__(
         """
         self.row_keys: set[bytes] = set()
         self.row_ranges: list[RowRange | dict[str, bytes]] = []
-        if row_ranges:
+        if row_ranges is not None:
             if isinstance(row_ranges, RowRange):
                 row_ranges = [row_ranges]
             for r in row_ranges:
                 self.add_range(r)
-        if row_keys:
+        if row_keys is not None:
             if not isinstance(row_keys, list):
                 row_keys = [row_keys]
             for k in row_keys:
@@ -221,7 +221,11 @@ def _to_dict(self) -> dict[str, Any]:
             row_ranges.append(dict_range)
         row_keys = list(self.row_keys)
         row_keys.sort()
-        row_set = {"row_keys": row_keys, "row_ranges": row_ranges}
+        row_set: dict[str, Any] = {}
+        if row_keys:
+            row_set["row_keys"] = row_keys
+        if row_ranges:
+            row_set["row_ranges"] = row_ranges
         final_dict: dict[str, Any] = {
             "rows": row_set,
         }

tests/system/test_system.py

@@ -158,7 +158,7 @@ def __init__(self, table):
         self.table = table
 
     async def add_row(
-        self, row_key, family=TEST_FAMILY, qualifier=b"q", value=b"test-value"
+        self, row_key, *, family=TEST_FAMILY, qualifier=b"q", value=b"test-value"
     ):
         if isinstance(value, str):
             value = value.encode("utf-8")
@@ -397,6 +397,56 @@ async def test_read_rows_stream_inactive_timer(table, temp_rows):
         assert "idle_timeout=0.1" in str(e)
 
 
+@pytest.mark.asyncio
+async def test_read_row(table, temp_rows):
+    """
+    Test read_row (single row helper)
+    """
+    from google.cloud.bigtable import Row
+
+    await temp_rows.add_row(b"row_key_1", value=b"value")
+    row = await table.read_row(b"row_key_1")
+    assert isinstance(row, Row)
+    assert row.row_key == b"row_key_1"
+    assert row.cells[0].value == b"value"
+
+
+@pytest.mark.asyncio
+async def test_read_row_missing(table):
+    """
+    Test read_row when row does not exist
+    """
+    from google.api_core import exceptions
+    from google.cloud.bigtable.exceptions import RowNotFound
+
+    row_key = "row_key_not_exist"
+    with pytest.raises(RowNotFound) as e:
+        await table.read_row(row_key)
+        assert str(e) == f"Row b'{row_key}' not found"
+    with pytest.raises(exceptions.InvalidArgument) as e:
+        await table.read_row("")
+        assert "Row kest must be non-empty" in str(e)
+
+
+@pytest.mark.asyncio
+async def test_row_exists(table, temp_rows):
+    from google.api_core import exceptions
+
+    """Test row_exists with rows that exist and don't exist"""
+    assert await table.row_exists(b"row_key_1") is False
+    await temp_rows.add_row(b"row_key_1")
+    assert await table.row_exists(b"row_key_1") is True
+    assert await table.row_exists("row_key_1") is True
+    assert await table.row_exists(b"row_key_2") is False
+    assert await table.row_exists("row_key_2") is False
+    assert await table.row_exists("3") is False
+    await temp_rows.add_row(b"3")
+    assert await table.row_exists(b"3") is True
+    with pytest.raises(exceptions.InvalidArgument) as e:
+        await table.row_exists("")
+        assert "Row kest must be non-empty" in str(e)
+
+
 @retry.Retry(predicate=retry.if_exception_type(ClientError), initial=1, maximum=5)
 @pytest.mark.parametrize(
     "cell_value,filter_input,expect_match",

tests/unit/test_client.py

@@ -1296,6 +1296,127 @@ async def test_read_rows_default_timeout_override(self):
                     assert kwargs["operation_timeout"] == operation_timeout
                     assert kwargs["per_request_timeout"] == per_request_timeout
 
+    @pytest.mark.asyncio
+    async def test_read_row(self):
+        """Test reading a single row"""
+        async with self._make_client() as client:
+            table = client.get_table("instance", "table")
+            row_key = b"test_1"
+            with mock.patch.object(table, "read_rows") as read_rows:
+                expected_result = object()
+                read_rows.side_effect = lambda *args, **kwargs: [expected_result]
+                expected_op_timeout = 8
+                expected_req_timeout = 4
+                row = await table.read_row(
+                    row_key,
+                    operation_timeout=expected_op_timeout,
+                    per_request_timeout=expected_req_timeout,
+                )
+                assert row == expected_result
+                assert read_rows.call_count == 1
+                args, kwargs = read_rows.call_args_list[0]
+                assert kwargs["operation_timeout"] == expected_op_timeout
+                assert kwargs["per_request_timeout"] == expected_req_timeout
+                assert len(args) == 1
+                assert isinstance(args[0], ReadRowsQuery)
+                assert args[0]._to_dict() == {
+                    "rows": {"row_keys": [row_key]},
+                    "rows_limit": 1,
+                }
+
+    @pytest.mark.asyncio
+    async def test_read_row_no_response(self):
+        """should raise RowNotFound if row does not exist"""
+        from google.cloud.bigtable.exceptions import RowNotFound
+
+        async with self._make_client() as client:
+            table = client.get_table("instance", "table")
+            row_key = b"test_1"
+            with mock.patch.object(table, "read_rows") as read_rows:
+                # return no rows
+                read_rows.side_effect = lambda *args, **kwargs: []
+                expected_op_timeout = 8
+                expected_req_timeout = 4
+                with pytest.raises(RowNotFound):
+                    await table.read_row(
+                        row_key,
+                        operation_timeout=expected_op_timeout,
+                        per_request_timeout=expected_req_timeout,
+                    )
+                assert read_rows.call_count == 1
+                args, kwargs = read_rows.call_args_list[0]
+                assert kwargs["operation_timeout"] == expected_op_timeout
+                assert kwargs["per_request_timeout"] == expected_req_timeout
+                assert isinstance(args[0], ReadRowsQuery)
+                assert args[0]._to_dict() == {
+                    "rows": {"row_keys": [row_key]},
+                    "rows_limit": 1,
+                }
+
+    @pytest.mark.parametrize("input_row", [None, 5, object()])
+    @pytest.mark.asyncio
+    async def test_read_row_w_invalid_input(self, input_row):
+        """Should raise error when passed None"""
+        async with self._make_client() as client:
+            table = client.get_table("instance", "table")
+            with pytest.raises(ValueError) as e:
+                await table.read_row(input_row)
+                assert "must be string or bytes" in e
+
+    @pytest.mark.parametrize(
+        "return_value,expected_result",
+        [
+            ([], False),
+            ([object()], True),
+            ([object(), object()], True),
+        ],
+    )
+    @pytest.mark.asyncio
+    async def test_row_exists(self, return_value, expected_result):
+        """Test checking for row existence"""
+        async with self._make_client() as client:
+            table = client.get_table("instance", "table")
+            row_key = b"test_1"
+            with mock.patch.object(table, "read_rows") as read_rows:
+                # return no rows
+                read_rows.side_effect = lambda *args, **kwargs: return_value
+                expected_op_timeout = 1
+                expected_req_timeout = 2
+                result = await table.row_exists(
+                    row_key,
+                    operation_timeout=expected_op_timeout,
+                    per_request_timeout=expected_req_timeout,
+                )
+                assert expected_result == result
+                assert read_rows.call_count == 1
+                args, kwargs = read_rows.call_args_list[0]
+                assert kwargs["operation_timeout"] == expected_op_timeout
+                assert kwargs["per_request_timeout"] == expected_req_timeout
+                assert isinstance(args[0], ReadRowsQuery)
+                expected_filter = {
+                    "chain": {
+                        "filters": [
+                            {"cells_per_row_limit_filter": 1},
+                            {"strip_value_transformer": True},
+                        ]
+                    }
+                }
+                assert args[0]._to_dict() == {
+                    "rows": {"row_keys": [row_key]},
+                    "rows_limit": 1,
+                    "filter": expected_filter,
+                }
+
+    @pytest.mark.parametrize("input_row", [None, 5, object()])
+    @pytest.mark.asyncio
+    async def test_row_exists_w_invalid_input(self, input_row):
+        """Should raise error when passed None"""
+        async with self._make_client() as client:
+            table = client.get_table("instance", "table")
+            with pytest.raises(ValueError) as e:
+                await table.row_exists(input_row)
+                assert "must be string or bytes" in e
+
     @pytest.mark.parametrize("include_app_profile", [True, False])
     @pytest.mark.asyncio
     async def test_read_rows_metadata(self, include_app_profile):

tests/unit/test_read_rows_query.py

@@ -300,7 +300,7 @@ def test_to_dict_rows_default(self):
         output = query._to_dict()
         self.assertTrue(isinstance(output, dict))
         self.assertEqual(len(output.keys()), 1)
-        expected = {"rows": {"row_keys": [], "row_ranges": []}}
+        expected = {"rows": {}}
         self.assertEqual(output, expected)
 
         request_proto = ReadRowsRequest(**output)
@@ -355,5 +355,10 @@ def test_to_dict_rows_populated(self):
         filter_proto = request_proto.filter
         self.assertEqual(filter_proto, row_filter._to_pb())
 
+    def test_empty_row_set(self):
+        """Empty strings should be treated as keys inputs"""
+        query = self._make_one(row_keys="")
+        self.assertEqual(query.row_keys, {b""})
+
     def test_shard(self):
         pass