From bf0683c8d111ef6dfc72698958f71dac567a7d72 Mon Sep 17 00:00:00 2001 From: Stephen Rosen Date: Tue, 8 Aug 2023 01:22:15 -0500 Subject: [PATCH 1/6] Add 'is_list_of' type guard to new '_guards' This is a generic check for objects against `list[T]`. e.g. `is_list_of(foo, str)` It's added to a new module for type-guard-like helpers, `_guards`. --- src/globus_sdk/_guards.py | 17 +++++++++++++++++ src/globus_sdk/exc/api.py | 12 +++++------- src/globus_sdk/exc/err_info.py | 18 +++++------------- .../auth_requirements_error/_validators.py | 8 +++++--- src/globus_sdk/services/flows/errors.py | 3 ++- src/globus_sdk/services/timer/errors.py | 13 ++++--------- 6 files changed, 38 insertions(+), 33 deletions(-) create mode 100644 src/globus_sdk/_guards.py diff --git a/src/globus_sdk/_guards.py b/src/globus_sdk/_guards.py new file mode 100644 index 000000000..6770c5740 --- /dev/null +++ b/src/globus_sdk/_guards.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +import sys +import typing as t + +if sys.version_info >= (3, 10): + from typing import TypeGuard +else: + from typing_extensions import TypeGuard + +T = t.TypeVar("T") + + +def is_list_of(data: t.Any, typ: type[T]) -> TypeGuard[list[T]]: + if isinstance(data, list): + return all(isinstance(item, typ) for item in data) + return False diff --git a/src/globus_sdk/exc/api.py b/src/globus_sdk/exc/api.py index 340bfed3c..2dc275896 100644 --- a/src/globus_sdk/exc/api.py +++ b/src/globus_sdk/exc/api.py @@ -6,6 +6,8 @@ import requests +from globus_sdk import _guards + from .base import GlobusError from .err_info import ErrorInfoContainer from .warnings import warn_deprecated @@ -273,7 +275,7 @@ def _detect_error_format(self) -> _ErrorFormat: # well-formed if self._jsonapi_mimetype(): errors = self._dict_data.get("errors") - if not isinstance(errors, list): + if not _guards.is_list_of(errors, dict): return _ErrorFormat.undefined elif len(errors) < 1: return _ErrorFormat.undefined @@ -321,9 +323,7 @@ def _parse_type_zero_error_format(self) -> bool: self.code = self._dict_data["code"] self.messages = [self._dict_data["message"]] self.request_id = self._dict_data.get("request_id") - if isinstance(self._dict_data.get("errors"), list) and all( - isinstance(subdoc, dict) for subdoc in self._dict_data["errors"] - ): + if _guards.is_list_of(self._dict_data.get("errors"), dict): raw_errors = self._dict_data["errors"] else: raw_errors = [self._dict_data] @@ -341,9 +341,7 @@ def _parse_undefined_error_format(self) -> bool: """ # attempt to pull out errors if possible and valid - if isinstance(self._dict_data.get("errors"), list) and all( - isinstance(subdoc, dict) for subdoc in self._dict_data["errors"] - ): + if _guards.is_list_of(self._dict_data.get("errors"), dict): raw_errors = self._dict_data["errors"] # if no 'errors' were found, or 'errors' is invalid, then # 'errors' should be set to contain the root document diff --git a/src/globus_sdk/exc/err_info.py b/src/globus_sdk/exc/err_info.py index 7e0bd002a..3ccc1a2d8 100644 --- a/src/globus_sdk/exc/err_info.py +++ b/src/globus_sdk/exc/err_info.py @@ -1,21 +1,13 @@ from __future__ import annotations import logging -import sys import typing as t -if sys.version_info < (3, 10): - from typing_extensions import TypeGuard -else: - from typing import TypeGuard +from globus_sdk import _guards log = logging.getLogger(__name__) -def _is_list_of_strs(obj: t.Any) -> TypeGuard[list[str]]: - return isinstance(obj, list) and all(isinstance(item, str) for item in obj) - - class ErrorInfo: """ Errors may contain "containers" of data which are testable (define ``__bool__``). @@ -98,7 +90,7 @@ def _parse_session_required_identities( self, data: dict[str, t.Any] ) -> list[str] | None: session_required_identities = data.get("session_required_identities") - if _is_list_of_strs(session_required_identities): + if _guards.is_list_of(session_required_identities, str): return session_required_identities elif session_required_identities is not None: self._warn_type( @@ -112,7 +104,7 @@ def _parse_session_required_single_domain( self, data: dict[str, t.Any] ) -> list[str] | None: session_required_single_domain = data.get("session_required_single_domain") - if _is_list_of_strs(session_required_single_domain): + if _guards.is_list_of(session_required_single_domain, str): return session_required_single_domain elif session_required_single_domain is not None: self._warn_type( @@ -128,7 +120,7 @@ def _parse_session_required_policies( session_required_policies = data.get("session_required_policies") if isinstance(session_required_policies, str): return session_required_policies.split(",") - elif _is_list_of_strs(session_required_policies): + elif _guards.is_list_of(session_required_policies, str): return session_required_policies elif session_required_policies is not None: self._warn_type( @@ -162,7 +154,7 @@ def __init__(self, error_data: dict[str, t.Any]) -> None: self._has_data = has_code and bool(self.required_scopes) def _parse_required_scopes(self, data: dict[str, t.Any]) -> list[str]: - if _is_list_of_strs(data.get("required_scopes")): + if _guards.is_list_of(data.get("required_scopes"), str): return t.cast("list[str]", data["required_scopes"]) elif isinstance(data.get("required_scope"), str): return [data["required_scope"]] diff --git a/src/globus_sdk/experimental/auth_requirements_error/_validators.py b/src/globus_sdk/experimental/auth_requirements_error/_validators.py index c1873e62d..49ae89ee0 100644 --- a/src/globus_sdk/experimental/auth_requirements_error/_validators.py +++ b/src/globus_sdk/experimental/auth_requirements_error/_validators.py @@ -2,6 +2,8 @@ import typing as t +from globus_sdk import _guards + from ._serializable import Serializable S = t.TypeVar("S", bound=Serializable) @@ -32,7 +34,7 @@ def opt_bool(name: str, value: t.Any) -> bool | None: def str_list(name: str, value: t.Any) -> list[str]: - if isinstance(value, list) and all(isinstance(s, str) for s in value): + if _guards.is_list_of(value, str): return value raise ValidationError(f"'{name}' must be a list of strings") @@ -40,7 +42,7 @@ def str_list(name: str, value: t.Any) -> list[str]: def opt_str_list(name: str, value: t.Any) -> list[str] | None: if value is None: return None - if isinstance(value, list) and all(isinstance(s, str) for s in value): + if _guards.is_list_of(value, str): return value raise ValidationError(f"'{name}' must be a list of strings or null") @@ -50,7 +52,7 @@ def opt_str_list_or_commasep(name: str, value: t.Any) -> list[str] | None: return None if isinstance(value, str): value = value.split(",") - if isinstance(value, list) and all(isinstance(s, str) for s in value): + if _guards.is_list_of(value, str): return value raise ValidationError( f"'{name}' must be a list of strings or a comma-delimited string or null" diff --git a/src/globus_sdk/services/flows/errors.py b/src/globus_sdk/services/flows/errors.py index e7ab08079..270a5adf0 100644 --- a/src/globus_sdk/services/flows/errors.py +++ b/src/globus_sdk/services/flows/errors.py @@ -1,5 +1,6 @@ from __future__ import annotations +from globus_sdk import _guards from globus_sdk.exc import ErrorSubdocument, GlobusAPIError @@ -27,7 +28,7 @@ def _parse_undefined_error_format(self) -> bool: self.code = self._extract_code_from_error_array(self.errors) details = self._dict_data["error"].get("detail") - if isinstance(details, list): + if _guards.is_list_of(details, dict): self.messages = [ error_detail["msg"] for error_detail in details diff --git a/src/globus_sdk/services/timer/errors.py b/src/globus_sdk/services/timer/errors.py index 5b04bcb8a..3482b602b 100644 --- a/src/globus_sdk/services/timer/errors.py +++ b/src/globus_sdk/services/timer/errors.py @@ -2,6 +2,7 @@ import typing as t +from globus_sdk import _guards from globus_sdk.exc import ErrorSubdocument, GlobusAPIError @@ -54,7 +55,7 @@ def _parse_undefined_error_format(self) -> bool: self.code = self._extract_code_from_error_array(self.errors) self.messages = self._extract_messages_from_error_array(self.errors) return True - elif isinstance(self._dict_data.get("detail"), list): + elif _guards.is_list_of(self._dict_data.get("detail"), dict): # FIXME: # the 'code' is currently being set explicitly by the # SDK in this case even though none was provided by @@ -63,11 +64,7 @@ def _parse_undefined_error_format(self) -> bool: self.code = "Validation Error" # collect the errors array from details - self.errors = [ - ErrorSubdocument(d) - for d in self._dict_data["detail"] - if isinstance(d, dict) - ] + self.errors = [ErrorSubdocument(d) for d in self._dict_data["detail"]] # drop error objects which don't have the relevant fields # and then build custom 'messages' for Globus Timers errors @@ -87,8 +84,6 @@ def _details_from_errors( if not isinstance(d.get("msg"), str): continue loc_list = d.get("loc") - if not isinstance(loc_list, list): - continue - if not all(isinstance(path_item, str) for path_item in loc_list): + if not _guards.is_list_of(loc_list, str): continue yield d.raw From f9ae9e5220ab56ed241b9f1a2129a0be8664ee9d Mon Sep 17 00:00:00 2001 From: Stephen Rosen Date: Tue, 8 Aug 2023 12:43:23 -0500 Subject: [PATCH 2/6] Introduce `_guards._is_optional{,_list_of}` These are internal helpers for very common dispatching isinstance checks. --- src/globus_sdk/_guards.py | 14 +++++++++++--- .../auth_requirements_error/_validators.py | 18 ++++++------------ src/globus_sdk/response.py | 4 +++- src/globus_sdk/services/auth/client/base.py | 5 ++--- 4 files changed, 22 insertions(+), 19 deletions(-) diff --git a/src/globus_sdk/_guards.py b/src/globus_sdk/_guards.py index 6770c5740..8d80e943e 100644 --- a/src/globus_sdk/_guards.py +++ b/src/globus_sdk/_guards.py @@ -12,6 +12,14 @@ def is_list_of(data: t.Any, typ: type[T]) -> TypeGuard[list[T]]: - if isinstance(data, list): - return all(isinstance(item, typ) for item in data) - return False + return isinstance(data, list) and all(isinstance(item, typ) for item in data) + + +def is_optional(data: t.Any, typ: type[T]) -> TypeGuard[T | None]: + return data is None or isinstance(data, typ) + + +def is_optional_list_of(data: t.Any, typ: type[T]) -> TypeGuard[list[T] | None]: + return data is None or ( + isinstance(data, list) and all(isinstance(item, typ) for item in data) + ) diff --git a/src/globus_sdk/experimental/auth_requirements_error/_validators.py b/src/globus_sdk/experimental/auth_requirements_error/_validators.py index 49ae89ee0..e14289f8a 100644 --- a/src/globus_sdk/experimental/auth_requirements_error/_validators.py +++ b/src/globus_sdk/experimental/auth_requirements_error/_validators.py @@ -20,15 +20,13 @@ def str_(name: str, value: t.Any) -> str: def opt_str(name: str, value: t.Any) -> str | None: - if value is None: - return None - if isinstance(value, str): + if _guards.is_optional(value, str): return value raise ValidationError(f"'{name}' must be a string or null") def opt_bool(name: str, value: t.Any) -> bool | None: - if value is None or isinstance(value, bool): + if _guards.is_optional(value, bool): return value raise ValidationError(f"'{name}' must be a bool or null") @@ -40,20 +38,16 @@ def str_list(name: str, value: t.Any) -> list[str]: def opt_str_list(name: str, value: t.Any) -> list[str] | None: - if value is None: - return None - if _guards.is_list_of(value, str): + if _guards.is_optional_list_of(value, str): return value raise ValidationError(f"'{name}' must be a list of strings or null") def opt_str_list_or_commasep(name: str, value: t.Any) -> list[str] | None: - if value is None: - return None - if isinstance(value, str): - value = value.split(",") - if _guards.is_list_of(value, str): + if _guards.is_optional_list_of(value, str): return value + if isinstance(value, str): + return value.split(",") raise ValidationError( f"'{name}' must be a list of strings or a comma-delimited string or null" ) diff --git a/src/globus_sdk/response.py b/src/globus_sdk/response.py index 0f144f78f..76dcf4e28 100644 --- a/src/globus_sdk/response.py +++ b/src/globus_sdk/response.py @@ -7,6 +7,8 @@ from requests import Response +from globus_sdk import _guards + log = logging.getLogger(__name__) if t.TYPE_CHECKING: @@ -134,7 +136,7 @@ def get(self, key: str, default: t.Any = None) -> t.Any: ``get`` is just an alias for ``data.get(key, default)``, but with the added checks that if ``data`` is ``None`` or a list, it returns the default. """ - if self.data is None or isinstance(self.data, list): + if _guards.is_optional(self.data, list): return default # NB: `default` is provided as a positional because the native dict type # doesn't recognize a keyword argument `default` diff --git a/src/globus_sdk/services/auth/client/base.py b/src/globus_sdk/services/auth/client/base.py index a5d713477..e4660b309 100644 --- a/src/globus_sdk/services/auth/client/base.py +++ b/src/globus_sdk/services/auth/client/base.py @@ -9,6 +9,7 @@ import jwt from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey +from globus_sdk import _guards from globus_sdk.authorizers import GlobusAuthorizer if sys.version_info >= (3, 8): @@ -701,9 +702,7 @@ def oauth2_validate_token( # if this client has no way of authenticating itself but # it does have a client_id, we'll send that in the request - no_authentication = self.authorizer is None or isinstance( - self.authorizer, NullAuthorizer - ) + no_authentication = _guards.is_optional(self.authorizer, NullAuthorizer) if no_authentication and self.client_id: log.debug("Validating token with unauthenticated client") body.update({"client_id": self.client_id}) From 24588aa63535cac87fba0b99e3762f27c9c22822 Mon Sep 17 00:00:00 2001 From: Stephen Rosen Date: Tue, 8 Aug 2023 13:24:15 -0500 Subject: [PATCH 3/6] Implement _guards.reduce for better chaining Validators in the experimental GARE subpackage can be built from TypeGuards if those type guards can be built in a form which is suitable to some kind of chaining API in which the type is an input and the guards are outputs or attributes. `reduce` is a purpose-built helper in `_guards` which provides a `_Reducer[T]` if given a `type[T]`. `reduce(str) -> _Reducer[str]`. The `_Reducer` can then have methods which act as guards of the desirable signatures. Finally, this can be pulled into the GARE component to write a TypeGuard -> ValidatorCallback helper and wrap the results of `reduce` calls. --- src/globus_sdk/_guards.py | 35 ++++++++++++++ .../auth_requirements_error/_validators.py | 47 +++++++++---------- 2 files changed, 57 insertions(+), 25 deletions(-) diff --git a/src/globus_sdk/_guards.py b/src/globus_sdk/_guards.py index 8d80e943e..6eaebcdbe 100644 --- a/src/globus_sdk/_guards.py +++ b/src/globus_sdk/_guards.py @@ -23,3 +23,38 @@ def is_optional_list_of(data: t.Any, typ: type[T]) -> TypeGuard[list[T] | None]: return data is None or ( isinstance(data, list) and all(isinstance(item, typ) for item in data) ) + + +def reduce(typ: type[T]) -> _Reducer[T]: + """ + reduce and the supporting _Reducer class provide a transform on existing type + guards, defined above. + + The basic transform is from a check over two parameters to a check over one + (input) parameter. + + e.g. The following two guards are the same: + is_list_of(data, str) + reduce(str).list_of(data) + + The effect is to produce a simpler call signature which is easier to chain + or otherwise modify. + """ + return _Reducer(typ) + + +class _Reducer(t.Generic[T]): + def __init__(self, typ: type[T]): + self.typ = typ + + def __call__(self, data: t.Any) -> TypeGuard[T]: + return isinstance(data, self.typ) + + def list_of(self, data: t.Any) -> TypeGuard[list[T]]: + return is_list_of(data, self.typ) + + def optional(self, data: t.Any) -> TypeGuard[T | None]: + return is_optional(data, self.typ) + + def optional_list(self, data: t.Any) -> TypeGuard[list[T] | None]: + return is_optional_list_of(data, self.typ) diff --git a/src/globus_sdk/experimental/auth_requirements_error/_validators.py b/src/globus_sdk/experimental/auth_requirements_error/_validators.py index e14289f8a..3d8f71973 100644 --- a/src/globus_sdk/experimental/auth_requirements_error/_validators.py +++ b/src/globus_sdk/experimental/auth_requirements_error/_validators.py @@ -1,11 +1,18 @@ from __future__ import annotations +import sys import typing as t from globus_sdk import _guards from ._serializable import Serializable +if sys.version_info >= (3, 10): + from typing import TypeGuard +else: + from typing_extensions import TypeGuard + +T = t.TypeVar("T") S = t.TypeVar("S", bound=Serializable) @@ -13,34 +20,24 @@ class ValidationError(ValueError): pass -def str_(name: str, value: t.Any) -> str: - if isinstance(value, str): - return value - raise ValidationError(f"'{name}' must be a string") +def _from_guard( + check: t.Callable[[t.Any], TypeGuard[T]], description: str +) -> t.Callable[[str, t.Any], T]: + def validator(name: str, value: t.Any) -> T: + if check(value): + return value + raise ValidationError(f"'{name}' must be {description}") + return validator -def opt_str(name: str, value: t.Any) -> str | None: - if _guards.is_optional(value, str): - return value - raise ValidationError(f"'{name}' must be a string or null") - - -def opt_bool(name: str, value: t.Any) -> bool | None: - if _guards.is_optional(value, bool): - return value - raise ValidationError(f"'{name}' must be a bool or null") - -def str_list(name: str, value: t.Any) -> list[str]: - if _guards.is_list_of(value, str): - return value - raise ValidationError(f"'{name}' must be a list of strings") - - -def opt_str_list(name: str, value: t.Any) -> list[str] | None: - if _guards.is_optional_list_of(value, str): - return value - raise ValidationError(f"'{name}' must be a list of strings or null") +str_ = _from_guard(_guards.reduce(str), "a string") +opt_str = _from_guard(_guards.reduce(str).optional, "a string or null") +opt_bool = _from_guard(_guards.reduce(bool).optional, "a bool or null") +str_list = _from_guard(_guards.reduce(str).list_of, "a list of strings") +opt_str_list = _from_guard( + _guards.reduce(str).optional_list, "a list of strings or null" +) def opt_str_list_or_commasep(name: str, value: t.Any) -> list[str] | None: From 0fb8c69c5a01a0986e04d456930dae1d7dfea869 Mon Sep 17 00:00:00 2001 From: Stephen Rosen Date: Tue, 8 Aug 2023 14:03:44 -0500 Subject: [PATCH 4/6] Add runtime and typing tests for _guards This also leads to a minor change from __call__ to "apply()" in order to ensure type-checkers read the types correctly. --- src/globus_sdk/_guards.py | 4 +- .../auth_requirements_error/_validators.py | 2 +- .../mypy-ignore-tests/test_guards.py | 51 +++++++++++ tests/unit/test_guards.py | 86 +++++++++++++++++++ 4 files changed, 141 insertions(+), 2 deletions(-) create mode 100644 tests/non-pytest/mypy-ignore-tests/test_guards.py create mode 100644 tests/unit/test_guards.py diff --git a/src/globus_sdk/_guards.py b/src/globus_sdk/_guards.py index 6eaebcdbe..d951e7e1c 100644 --- a/src/globus_sdk/_guards.py +++ b/src/globus_sdk/_guards.py @@ -47,7 +47,9 @@ class _Reducer(t.Generic[T]): def __init__(self, typ: type[T]): self.typ = typ - def __call__(self, data: t.Any) -> TypeGuard[T]: + # although it might be nice to use __call__ here, type checkers (read: mypy) do not + # always correctly treat an arbitrary callable TypeGuard as a a guard + def apply(self, data: t.Any) -> TypeGuard[T]: return isinstance(data, self.typ) def list_of(self, data: t.Any) -> TypeGuard[list[T]]: diff --git a/src/globus_sdk/experimental/auth_requirements_error/_validators.py b/src/globus_sdk/experimental/auth_requirements_error/_validators.py index 3d8f71973..3d24660b6 100644 --- a/src/globus_sdk/experimental/auth_requirements_error/_validators.py +++ b/src/globus_sdk/experimental/auth_requirements_error/_validators.py @@ -31,7 +31,7 @@ def validator(name: str, value: t.Any) -> T: return validator -str_ = _from_guard(_guards.reduce(str), "a string") +str_ = _from_guard(_guards.reduce(str).apply, "a string") opt_str = _from_guard(_guards.reduce(str).optional, "a string or null") opt_bool = _from_guard(_guards.reduce(bool).optional, "a bool or null") str_list = _from_guard(_guards.reduce(str).list_of, "a list of strings") diff --git a/tests/non-pytest/mypy-ignore-tests/test_guards.py b/tests/non-pytest/mypy-ignore-tests/test_guards.py new file mode 100644 index 000000000..ce6a4070b --- /dev/null +++ b/tests/non-pytest/mypy-ignore-tests/test_guards.py @@ -0,0 +1,51 @@ +# test that the internal _guards module provides valid and well-formed type-guards +import typing as t + +from globus_sdk import _guards + + +def get_any() -> t.Any: + return 1 + + +x = get_any() +t.assert_type(x, t.Any) + +# test reduce().apply +if _guards.reduce(str).apply(x): + y1 = x + t.assert_type(y1, str) + +# test is_list_of / reduce().list_of +if _guards.is_list_of(x, str): + t.assert_type(x, list[str]) +elif _guards.is_list_of(x, int): + t.assert_type(x, list[int]) + +if _guards.reduce(str).list_of(x): + t.assert_type(x, list[str]) +elif _guards.reduce(int).list_of(x): + t.assert_type(x, list[int]) + +# test is_optional / reduce().optional +if _guards.is_optional(x, float): + t.assert_type(x, float | None) +elif _guards.is_optional(x, bytes): + t.assert_type(x, bytes | None) + +if _guards.reduce(float).optional(x): + t.assert_type(x, float | None) +elif _guards.reduce(bytes).optional(x): + t.assert_type(x, bytes | None) + + +# test is_optional_list_of / reduce().optional_list +if _guards.is_optional_list_of(x, type(None)): + t.assert_type(x, list[None] | None) +elif _guards.is_optional_list_of(x, dict): + t.assert_type(x, list[dict[t.Any, t.Any]] | None) + +if _guards.reduce(type(None)).optional_list(x): + t.assert_type(x, list[None] | None) +elif _guards.reduce(dict).optional_list(x): + t.assert_type(x, list[dict[t.Any, t.Any]] | None) diff --git a/tests/unit/test_guards.py b/tests/unit/test_guards.py new file mode 100644 index 000000000..6dd16ccda --- /dev/null +++ b/tests/unit/test_guards.py @@ -0,0 +1,86 @@ +import pytest + +from globus_sdk import _guards + + +@pytest.mark.parametrize( + "value, typ, ok", + [ + # passing + ([], str, True), + ([1, 2], int, True), + (["1", ""], str, True), + ([], list, True), + ([[], [1, 2], ["foo"]], list, True), + # failing + ([1], str, False), + (["foo"], int, False), + ((1, 2), int, False), + (list, list, False), + (list, str, False), + (["foo", 1], str, False), + ([1, 2], list, False), + ], +) +def test_list_of_guard(value, typ, ok): + assert _guards.is_list_of(value, typ) == ok + + +@pytest.mark.parametrize( + "value, typ, ok", + [ + # passing + (None, str, True), + ("foo", str, True), + # failing + (b"foo", str, False), + ("", int, False), + (type(None), str, False), + ], +) +def test_opt_guard(value, typ, ok): + assert _guards.is_optional(value, typ) == ok + + +@pytest.mark.parametrize( + "value, typ, ok", + [ + # passing + ([], str, True), + ([], int, True), + ([1, 2], int, True), + (["1", ""], str, True), + (None, str, True), + # failing + # NB: the guard checks `list[str] | None`, not `list[str | None]` + ([None], str, False), + (b"foo", str, False), + ("", str, False), + (type(None), str, False), + ], +) +def test_opt_list_guard(value, typ, ok): + assert _guards.is_optional_list_of(value, typ) == ok + + +def test_reduced_guards(): + reduced_str = _guards.reduce(str) + + # apply => isinstance + assert reduced_str.apply("foo") + assert not reduced_str.apply(None) + + # list_of => is_list_of + assert reduced_str.list_of(["foo", "bar"]) + assert not reduced_str.list_of(["foo", "bar", 1]) + + # optional => is_optional + assert reduced_str.optional("foo") + assert reduced_str.optional(None) + assert not reduced_str.optional(1) + + # optional_list => is_optional_list_of + assert reduced_str.optional_list(["foo", "bar"]) + assert not reduced_str.optional_list(["foo", "bar", 1]) + assert reduced_str.optional_list(None) + assert not reduced_str.optional_list("") From c967fd96662259e50155c34a917a6b05f1009e66 Mon Sep 17 00:00:00 2001 From: Stephen Rosen Date: Thu, 10 Aug 2023 16:58:07 -0500 Subject: [PATCH 5/6] Revert "Implement _guards.reduce for better chaining" This reverts commit 24588aa63535cac87fba0b99e3762f27c9c22822. --- src/globus_sdk/_guards.py | 37 --------------- .../auth_requirements_error/_validators.py | 47 ++++++++++--------- 2 files changed, 25 insertions(+), 59 deletions(-) diff --git a/src/globus_sdk/_guards.py b/src/globus_sdk/_guards.py index d951e7e1c..8d80e943e 100644 --- a/src/globus_sdk/_guards.py +++ b/src/globus_sdk/_guards.py @@ -23,40 +23,3 @@ def is_optional_list_of(data: t.Any, typ: type[T]) -> TypeGuard[list[T] | None]: return data is None or ( isinstance(data, list) and all(isinstance(item, typ) for item in data) ) - - -def reduce(typ: type[T]) -> _Reducer[T]: - """ - reduce and the supporting _Reducer class provide a transform on existing type - guards, defined above. - - The basic transform is from a check over two parameters to a check over one - (input) parameter. - - e.g. The following two guards are the same: - is_list_of(data, str) - reduce(str).list_of(data) - - The effect is to produce a simpler call signature which is easier to chain - or otherwise modify. - """ - return _Reducer(typ) - - -class _Reducer(t.Generic[T]): - def __init__(self, typ: type[T]): - self.typ = typ - - # although it might be nice to use __call__ here, type checkers (read: mypy) do not - # always correctly treat an arbitrary callable TypeGuard as a a guard - def apply(self, data: t.Any) -> TypeGuard[T]: - return isinstance(data, self.typ) - - def list_of(self, data: t.Any) -> TypeGuard[list[T]]: - return is_list_of(data, self.typ) - - def optional(self, data: t.Any) -> TypeGuard[T | None]: - return is_optional(data, self.typ) - - def optional_list(self, data: t.Any) -> TypeGuard[list[T] | None]: - return is_optional_list_of(data, self.typ) diff --git a/src/globus_sdk/experimental/auth_requirements_error/_validators.py b/src/globus_sdk/experimental/auth_requirements_error/_validators.py index 3d24660b6..e14289f8a 100644 --- a/src/globus_sdk/experimental/auth_requirements_error/_validators.py +++ b/src/globus_sdk/experimental/auth_requirements_error/_validators.py @@ -1,18 +1,11 @@ from __future__ import annotations -import sys import typing as t from globus_sdk import _guards from ._serializable import Serializable -if sys.version_info >= (3, 10): - from typing import TypeGuard -else: - from typing_extensions import TypeGuard - -T = t.TypeVar("T") S = t.TypeVar("S", bound=Serializable) @@ -20,24 +13,34 @@ class ValidationError(ValueError): pass -def _from_guard( - check: t.Callable[[t.Any], TypeGuard[T]], description: str -) -> t.Callable[[str, t.Any], T]: - def validator(name: str, value: t.Any) -> T: - if check(value): - return value - raise ValidationError(f"'{name}' must be {description}") +def str_(name: str, value: t.Any) -> str: + if isinstance(value, str): + return value + raise ValidationError(f"'{name}' must be a string") - return validator +def opt_str(name: str, value: t.Any) -> str | None: + if _guards.is_optional(value, str): + return value + raise ValidationError(f"'{name}' must be a string or null") + + +def opt_bool(name: str, value: t.Any) -> bool | None: + if _guards.is_optional(value, bool): + return value + raise ValidationError(f"'{name}' must be a bool or null") -str_ = _from_guard(_guards.reduce(str).apply, "a string") -opt_str = _from_guard(_guards.reduce(str).optional, "a string or null") -opt_bool = _from_guard(_guards.reduce(bool).optional, "a bool or null") -str_list = _from_guard(_guards.reduce(str).list_of, "a list of strings") -opt_str_list = _from_guard( - _guards.reduce(str).optional_list, "a list of strings or null" -) + +def str_list(name: str, value: t.Any) -> list[str]: + if _guards.is_list_of(value, str): + return value + raise ValidationError(f"'{name}' must be a list of strings") + + +def opt_str_list(name: str, value: t.Any) -> list[str] | None: + if _guards.is_optional_list_of(value, str): + return value + raise ValidationError(f"'{name}' must be a list of strings or null") def opt_str_list_or_commasep(name: str, value: t.Any) -> list[str] | None: From dca7298afa8b735754c89e03857fb0f16dba82b7 Mon Sep 17 00:00:00 2001 From: Stephen Rosen Date: Thu, 10 Aug 2023 16:58:34 -0500 Subject: [PATCH 6/6] Remove tests for `_guards.reduce` --- .../mypy-ignore-tests/test_guards.py | 26 +++---------------- tests/unit/test_guards.py | 23 ---------------- 2 files changed, 3 insertions(+), 46 deletions(-) diff --git a/tests/non-pytest/mypy-ignore-tests/test_guards.py b/tests/non-pytest/mypy-ignore-tests/test_guards.py index ce6a4070b..0dc761533 100644 --- a/tests/non-pytest/mypy-ignore-tests/test_guards.py +++ b/tests/non-pytest/mypy-ignore-tests/test_guards.py @@ -11,41 +11,21 @@ def get_any() -> t.Any: x = get_any() t.assert_type(x, t.Any) -# test reduce().apply -if _guards.reduce(str).apply(x): - y1 = x - t.assert_type(y1, str) - -# test is_list_of / reduce().list_of +# test is_list_of if _guards.is_list_of(x, str): t.assert_type(x, list[str]) elif _guards.is_list_of(x, int): t.assert_type(x, list[int]) -if _guards.reduce(str).list_of(x): - t.assert_type(x, list[str]) -elif _guards.reduce(int).list_of(x): - t.assert_type(x, list[int]) - -# test is_optional / reduce().optional +# test is_optional if _guards.is_optional(x, float): t.assert_type(x, float | None) elif _guards.is_optional(x, bytes): t.assert_type(x, bytes | None) -if _guards.reduce(float).optional(x): - t.assert_type(x, float | None) -elif _guards.reduce(bytes).optional(x): - t.assert_type(x, bytes | None) - -# test is_optional_list_of / reduce().optional_list +# test is_optional_list_of if _guards.is_optional_list_of(x, type(None)): t.assert_type(x, list[None] | None) elif _guards.is_optional_list_of(x, dict): t.assert_type(x, list[dict[t.Any, t.Any]] | None) - -if _guards.reduce(type(None)).optional_list(x): - t.assert_type(x, list[None] | None) -elif _guards.reduce(dict).optional_list(x): - t.assert_type(x, list[dict[t.Any, t.Any]] | None) diff --git a/tests/unit/test_guards.py b/tests/unit/test_guards.py index 6dd16ccda..8b3b4d642 100644 --- a/tests/unit/test_guards.py +++ b/tests/unit/test_guards.py @@ -61,26 +61,3 @@ def test_opt_guard(value, typ, ok): ) def test_opt_list_guard(value, typ, ok): assert _guards.is_optional_list_of(value, typ) == ok - - -def test_reduced_guards(): - reduced_str = _guards.reduce(str) - - # apply => isinstance - assert reduced_str.apply("foo") - assert not reduced_str.apply(None) - - # list_of => is_list_of - assert reduced_str.list_of(["foo", "bar"]) - assert not reduced_str.list_of(["foo", "bar", 1]) - - # optional => is_optional - assert reduced_str.optional("foo") - assert reduced_str.optional(None) - assert not reduced_str.optional(1) - - # optional_list => is_optional_list_of - assert reduced_str.optional_list(["foo", "bar"]) - assert not reduced_str.optional_list(["foo", "bar", 1]) - assert reduced_str.optional_list(None) - assert not reduced_str.optional_list("")