feat: adds new input validation function similar to isinstance.

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)

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)