Raise a warning on unsafe/lossy cast to int, float, and bools (#242)

* Raise a warning on lossy int casting when loading

- Fixes #227: When the decoding function for ints would cast to an int
in a way that loses precision (e.g. from a float), then we raise a
warning.

Signed-off-by: Fabrice Normandin <normandf@mila.quebec>

* Broaden test, reduce # of warnings

Signed-off-by: Fabrice Normandin <normandf@mila.quebec>

* Add decoding_fn_for_type decorator, improve test

Signed-off-by: Fabrice Normandin <normandf@mila.quebec>

* Fix type annotation, add docstring in other fn

Signed-off-by: Fabrice Normandin <normandf@mila.quebec>

* Fix typing issue with py38

Signed-off-by: Fabrice Normandin <normandf@mila.quebec>

---------

Signed-off-by: Fabrice Normandin <normandf@mila.quebec>

Author: lebrice

simple_parsing/helpers/serialization/decoding.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import inspect
+import sys
 import warnings
 from collections import OrderedDict
 from collections.abc import Mapping
@@ -12,7 +13,6 @@
 from logging import getLogger
 from pathlib import Path
 from typing import Any, Callable, TypeVar
-from typing_extensions import Literal
 
 from simple_parsing.annotation_utils.get_field_annotations import (
     evaluate_string_annotation,
@@ -44,17 +44,66 @@
 _decoding_fns: dict[type[T], Callable[[Any], T]] = {
     # the 'primitive' types are decoded using the type fn as a constructor.
     t: t
-    for t in [str, float, int, bytes]
+    for t in [str, bytes]
 }
 
 
-def decode_bool(v: Any) -> bool:
-    if isinstance(v, str):
-        return str2bool(v)
-    return bool(v)
+def register_decoding_fn(
+    some_type: type[T], function: Callable[[Any], T], overwrite: bool = False
+) -> None:
+    """Register a decoding function for the type `some_type`."""
+    _register(some_type, function, overwrite=overwrite)
+
+
+def _register(t: type, func: Callable, overwrite: bool = False) -> None:
+    if t not in _decoding_fns or overwrite:
+        # logger.debug(f"Registering the type {t} with decoding function {func}")
+        _decoding_fns[t] = func
+
 
+C = TypeVar("C", bound=Callable[[Any], Any])
 
-_decoding_fns[bool] = decode_bool
+
+def decoding_fn_for_type(some_type: type) -> Callable[[C], C]:
+    """Registers a function to be used to convert a serialized value to the given type.
+
+    The function should accept one argument (the serialized value) and return the decoded value.
+    """
+
+    def _wrapper(fn: C) -> C:
+        register_decoding_fn(some_type, fn, overwrite=True)
+        return fn
+
+    return _wrapper
+
+
+@decoding_fn_for_type(int)
+def _decode_int(v: str) -> int:
+    int_v = int(v)
+    if isinstance(v, bool):
+        warnings.warn(UnsafeCastingWarning(raw_value=v, decoded_value=int_v))
+    elif int_v != float(v):
+        warnings.warn(UnsafeCastingWarning(raw_value=v, decoded_value=int_v))
+    return int_v
+
+
+@decoding_fn_for_type(float)
+def _decode_float(v: Any) -> float:
+    float_v = float(v)
+    if isinstance(v, bool):
+        warnings.warn(UnsafeCastingWarning(raw_value=v, decoded_value=float_v))
+    return float_v
+
+
+@decoding_fn_for_type(bool)
+def _decode_bool(v: Any) -> bool:
+    if isinstance(v, str):
+        bool_v = str2bool(v)
+    else:
+        bool_v = bool(v)
+        if isinstance(v, (int, float)) and v not in (0, 1, 0.0, 1.0):
+            warnings.warn(UnsafeCastingWarning(raw_value=v, decoded_value=bool_v))
+    return bool_v
 
 
 def decode_field(
@@ -93,11 +142,36 @@ def decode_field(
 
     decoding_function = get_decoding_fn(field_type)
 
-    if is_dataclass_type(field_type) and drop_extra_fields is not None:
-        # Pass the drop_extra_fields argument to the decoding function.
-        return decoding_function(raw_value, drop_extra_fields=drop_extra_fields)
+    _kwargs = dict(category=UnsafeCastingWarning) if sys.version_info >= (3, 11) else {}
 
-    return decoding_function(raw_value)
+    with warnings.catch_warnings(record=True, **_kwargs) as warning_messages:
+        if is_dataclass_type(field_type) and drop_extra_fields is not None:
+            # Pass the drop_extra_fields argument to the decoding function.
+            decoded_value = decoding_function(raw_value, drop_extra_fields=drop_extra_fields)
+        else:
+            decoded_value = decoding_function(raw_value)
+
+    for warning_message in warning_messages.copy():
+        if not isinstance(warning_message.message, UnsafeCastingWarning):
+            warnings.warn_explicit(
+                message=warning_message.message,
+                category=warning_message.category,
+                filename=warning_message.filename,
+                lineno=warning_message.lineno,
+                # module=warning_message.module,
+                # registry=warning_message.registry,
+                # module_globals=warning_message.module_globals,
+            )
+            warning_messages.remove(warning_message)
+
+    if warning_messages:
+        warnings.warn(
+            RuntimeWarning(
+                f"Unsafe casting occurred when deserializing field '{name}' of type {field_type}: "
+                f"raw value: {raw_value!r}, decoded value: {decoded_value!r}."
+            )
+        )
+    return decoded_value
 
 
 # NOTE: Disabling the caching here might help avoid some bugs, and it's unclear if this has that
@@ -224,7 +298,7 @@ def get_decoding_fn(type_annotation: type[T] | str) -> Callable[..., T]:
         logger.debug(f"Decoding a typevar: {t}, bound type is {bound}.")
         if bound is not None:
             return get_decoding_fn(bound)
-    
+
     if is_literal(t):
         logger.debug(f"Decoding a Literal field: {t}")
         possible_vals = get_type_arguments(t)
@@ -241,19 +315,6 @@ def get_decoding_fn(type_annotation: type[T] | str) -> Callable[..., T]:
     return try_constructor(t)
 
 
-def _register(t: type, func: Callable, overwrite: bool = False) -> None:
-    if t not in _decoding_fns or overwrite:
-        # logger.debug(f"Registering the type {t} with decoding function {func}")
-        _decoding_fns[t] = func
-
-
-def register_decoding_fn(
-    some_type: type[T], function: Callable[[Any], T], overwrite: bool = False
-) -> None:
-    """Register a decoding function for the type `some_type`."""
-    _register(some_type, function, overwrite=overwrite)
-
-
 def decode_optional(t: type[T]) -> Callable[[Any | None], T | None]:
     decode = get_decoding_fn(t)
 
@@ -281,15 +342,21 @@ def _try_functions(val: Any) -> T | Any:
 
 
 def decode_union(*types: type[T]) -> Callable[[Any], T | Any]:
-    types = list(types)
-    optional = type(None) in types
+    types_list = list(types)
+    optional = type(None) in types_list
+
     # Partition the Union into None and non-None types.
-    while type(None) in types:
-        types.remove(type(None))
+    while type(None) in types_list:
+        types_list.remove(type(None))
 
     decoding_fns: list[Callable[[Any], T]] = [
-        decode_optional(t) if optional else get_decoding_fn(t) for t in types
+        decode_optional(t) if optional else get_decoding_fn(t) for t in types_list
     ]
+
+    # TODO: We could be a bit smarter about the order in which we try the functions, but for now,
+    # we just try the functions in the same order as the annotation, and return the result from the
+    # first function that doesn't raise an exception.
+
     # Try using each of the non-None types, in succession. Worst case, return the value.
     return try_functions(*decoding_fns)
 
@@ -455,3 +522,10 @@ def constructor(val):
 
 
 register_decoding_fn(Path, Path)
+
+
+class UnsafeCastingWarning(RuntimeWarning):
+    def __init__(self, raw_value: Any, decoded_value: Any) -> None:
+        super().__init__()
+        self.raw_value = raw_value
+        self.decoded_value = decoded_value

simple_parsing/utils.py

@@ -917,7 +917,13 @@ def unflatten(flattened: Mapping[tuple[K, ...], V]) -> PossiblyNestedDict[K, V]:
 
 
 def flatten_join(nested: PossiblyNestedMapping[str, V], sep: str = ".") -> dict[str, V]:
-    """Flatten a dictionary of dictionaries. When collisions occur, joins the keys with `sep`."""
+    """Flatten a dictionary of dictionaries. Joins different nesting levels with `sep` as separator.
+
+    >>> flatten_join({'a': {'b': 2, 'c': 3}, 'c': {'d': 3, 'e': 4}})
+    {'a.b': 2, 'a.c': 3, 'c.d': 3, 'c.e': 4}
+    >>> flatten_join({'a': {'b': 2, 'c': 3}, 'c': {'d': 3, 'e': 4}}, sep="/")
+    {'a/b': 2, 'a/c': 3, 'c/d': 3, 'c/e': 4}
+    """
     return {sep.join(keys): value for keys, value in flatten(nested).items()}
 
 

test/conftest.py

@@ -4,6 +4,7 @@
 import os
 import pathlib
 import sys
+import warnings
 from logging import getLogger as get_logger
 from typing import Any, Generic, TypeVar
 
@@ -173,8 +174,15 @@ def no_stdout(capsys, caplog):
         pytest.fail(f"Test generated some output in stderr: '{captured.err}'")
 
 
+@pytest.fixture(autouse=False)
+def no_warnings_raised():
+    with warnings.catch_warnings(record=True):
+        warnings.simplefilter("error")
+        yield
+
+
 @pytest.fixture
-def no_warnings(caplog):
+def no_warning_log_messages(caplog):
     yield
     for when in ("setup", "call"):
         messages = [x.message for x in caplog.get_records(when) if x.levelno == logging.WARNING]
@@ -183,7 +191,7 @@ def no_warnings(caplog):
 
 
 @pytest.fixture
-def silent(no_stdout, no_warnings):
+def silent(no_stdout, no_warning_log_messages):
     """
     Test fixture that will make a test fail if it prints anything to stdout or
     logs warnings

test/helpers/test_enum_serialization.py

@@ -31,9 +31,9 @@ class Parameters(Serializable):
 @pytest.mark.xfail(
     raises=KeyError, match="'jsonl'", strict=True, reason="Enums are saved by name, not by value."
 )
-def test_decode_enum_saved_by_value_doesnt_work():
+def test_decode_enum_saved_by_value_doesnt_work(tmp_path: Path):
     """Test to reproduce https://github.com/lebrice/SimpleParsing/issues/219#issuecomment-1437817369"""
-    with open("conf.yaml", "w") as f:
+    with open(tmp_path / "conf.yaml", "w") as f:
         f.write(
             textwrap.dedent(
                 """\
@@ -45,7 +45,7 @@ def test_decode_enum_saved_by_value_doesnt_work():
             )
         )
 
-    file_config = Parameters.load_yaml("conf.yaml")
+    file_config = Parameters.load_yaml(tmp_path / "conf.yaml")
     assert file_config == Parameters(hparams=Hparams(xyz=[LoggingTypes.JSONL]), p=Path("/tmp"))