feat: adds new input validation function similar to isinstance. (#2107)

chalmerlowe · gcf-owl-bot[bot] · web-flow · commit a2bebb95c5ef · 2025-01-09T13:42:37.000-05:00
* feat: adds new function similar to isinstance. * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
diff --git a/google/cloud/bigquery/_helpers.py b/google/cloud/bigquery/_helpers.py
@@ -22,7 +22,7 @@
 import re
 import os
 import warnings
-from typing import Optional, Union
+from typing import Optional, Union, Any, Tuple, Type
 
 from dateutil import relativedelta
 from google.cloud._helpers import UTC  # type: ignore
@@ -1004,3 +1004,33 @@ def _verify_job_config_type(job_config, expected_type, param_name="job_config"):
                 job_config=job_config,
             )
         )
+
+
+def _isinstance_or_raise(
+    value: Any,
+    dtype: Union[Type, Tuple[Type, ...]],
+    none_allowed: Optional[bool] = False,
+) -> Any:
+    """Determine whether a value type matches a given datatype or None.
+    Args:
+        value (Any): Value to be checked.
+        dtype (type): Expected data type or tuple of data types.
+        none_allowed Optional(bool): whether value is allowed to be None. Default
+           is False.
+    Returns:
+        Any: Returns the input value if the type check is successful.
+    Raises:
+        TypeError: If the input value's type does not match the expected data type(s).
+    """
+    if none_allowed and value is None:
+        return value
+
+    if isinstance(value, dtype):
+        return value
+
+    or_none = ""
+    if none_allowed:
+        or_none = " (or None)"
+
+    msg = f"Pass {value} as a '{dtype}'{or_none}. Got {type(value)}."
+    raise TypeError(msg)
diff --git a/tests/unit/test__helpers.py b/tests/unit/test__helpers.py
@@ -24,6 +24,7 @@
 from unittest import mock
 
 import google.api_core
+from google.cloud.bigquery._helpers import _isinstance_or_raise
 
 
 @pytest.mark.skipif(
@@ -1661,3 +1662,34 @@ def test_w_env_var(self):
             host = self._call_fut()
 
         self.assertEqual(host, HOST)
+
+
+class Test__isinstance_or_raise:
+    @pytest.mark.parametrize(
+        "value,dtype,none_allowed,expected",
+        [
+            (None, str, True, None),
+            ("hello world.uri", str, True, "hello world.uri"),
+            ("hello world.uri", str, False, "hello world.uri"),
+            (None, (str, float), True, None),
+            ("hello world.uri", (str, float), True, "hello world.uri"),
+            ("hello world.uri", (str, float), False, "hello world.uri"),
+        ],
+    )
+    def test__valid_isinstance_or_raise(self, value, dtype, none_allowed, expected):
+        result = _isinstance_or_raise(value, dtype, none_allowed=none_allowed)
+        assert result == expected
+
+    @pytest.mark.parametrize(
+        "value,dtype,none_allowed,expected",
+        [
+            (None, str, False, pytest.raises(TypeError)),
+            ({"key": "value"}, str, True, pytest.raises(TypeError)),
+            ({"key": "value"}, str, False, pytest.raises(TypeError)),
+            ({"key": "value"}, (str, float), True, pytest.raises(TypeError)),
+            ({"key": "value"}, (str, float), False, pytest.raises(TypeError)),
+        ],
+    )
+    def test__invalid_isinstance_or_raise(self, value, dtype, none_allowed, expected):
+        with expected:
+            _isinstance_or_raise(value, dtype, none_allowed=none_allowed)
