diff --git a/src/lob_hlpr/hlpr.py b/src/lob_hlpr/hlpr.py index ba316fb..5c95f41 100644 --- a/src/lob_hlpr/hlpr.py +++ b/src/lob_hlpr/hlpr.py @@ -1,10 +1,12 @@ import binascii +import json import logging import logging.handlers import os import re -from dataclasses import asdict +from dataclasses import fields, is_dataclass from datetime import datetime +from typing import Any from lob_hlpr.lib_types import FirmwareID @@ -165,65 +167,71 @@ def lob_print(log_path: str, *args, **kwargs): logger.info(*args, **kwargs) @staticmethod - def ascleandict(dclass, remove_false=False): + def ascleandict( + dclass: object, + remove_false: bool = False, + json_serializable: bool = False, + ) -> dict[str, Any]: """Convert a dataclass to a dictionary and remove None values. - Largely generated by AI... - Args: dclass: The dataclass instance to convert. remove_false: If True, also remove boolean fields that are False. + json_serializable: If True, convert all non-serializable types to strings. Returns: dict: The cleaned dictionary without empty values. + + Raises: + TypeError: If dclass is not a dataclass instance. """ + if not is_dataclass(dclass) or isinstance(dclass, type): + raise TypeError( + f"ascleandict() should be called on dataclass instances, " + f"not {type(dclass)!r}" + ) - def clean_value(v): - """Recursively clean nested structures.""" - if isinstance(v, dict): - cleaned = { - k: clean_value(val) - for (k, val) in v.items() - if (val is not None) - and not (isinstance(val, list) and len(val) == 0) - and not (isinstance(val, dict) and len(val) == 0) - and not (remove_false and (isinstance(val, bool) and val is False)) - } - # Keep removing empty dicts/lists until nothing changes - while True: - filtered = { - k: v - for (k, v) in cleaned.items() - if not (isinstance(v, dict) and len(v) == 0) - and not (isinstance(v, list) and len(v) == 0) - } - if len(filtered) == len(cleaned): - break - cleaned = filtered - return cleaned - elif isinstance(v, list): - cleaned_items = [clean_value(item) for item in v] - # Filter out empty dicts and lists from the result - return [ - item - for item in cleaned_items - if not (isinstance(item, dict) and len(item) == 0) - and not (isinstance(item, list) and len(item) == 0) - ] - return v - - result = asdict( - dclass, - dict_factory=lambda x: { - k: v - for (k, v) in x - if (v is not None) - and not (isinstance(v, list) and len(v) == 0) - and not (isinstance(v, dict) and len(v) == 0) - and not (remove_false and (isinstance(v, bool) and v is False)) - }, - ) - return clean_value(result) + def _keep(v) -> bool: + if v is None: + return False + if isinstance(v, (list, dict)) and not v: + return False + if remove_false and v is False: + return False + return True + + def _convert(obj: object) -> Any: + if is_dataclass(obj) and not isinstance(obj, type): + result = {} + for f in fields(obj): + converted = _convert(getattr(obj, f.name)) + if _keep(converted): + result[f.name] = converted + return result + if isinstance(obj, dict): + result = {} + for k, v in obj.items(): + converted = _convert(v) + if _keep(converted): + key = ( + str(k) + if json_serializable and not isinstance(k, str) + else k + ) + result[key] = converted + return result + if isinstance(obj, (list, tuple)): + items = [_convert(item) for item in obj] + items = [item for item in items if _keep(item)] + return tuple(items) if isinstance(obj, tuple) else items + if json_serializable: + try: + json.dumps(obj) + except (TypeError, OverflowError): + return str(obj) + return obj + + return _convert(dclass) @staticmethod def unix_timestamp() -> int: diff --git a/tests/test_lob_hlpr.py b/tests/test_lob_hlpr.py index d68d175..9abbdc3 100644 --- a/tests/test_lob_hlpr.py +++ b/tests/test_lob_hlpr.py @@ -1,3 +1,4 @@ +import json import logging from dataclasses import dataclass @@ -160,6 +161,21 @@ def test_log_print_passes(tmp_path, capsys): assert log_content.count("Another test message") == 1 +def test_ascleandict_rejects_non_dataclass(): + """Test that ascleandict raises TypeError for non-dataclass inputs.""" + with pytest.raises(TypeError): + hlp.ascleandict({"key": "value"}) + with pytest.raises(TypeError): + hlp.ascleandict([1, 2, 3]) + with pytest.raises(TypeError): + hlp.ascleandict("string") + with pytest.raises(TypeError): + hlp.ascleandict(42) + # Dataclass type (not instance) must also be rejected + with pytest.raises(TypeError): + hlp.ascleandict(dataclass) + + def test_as_clean_dict_passes(): """Test as_clean_dict function with valid inputs.""" @@ -210,6 +226,95 @@ class DataWithList: assert result2 == {"name": "test", "items": [{"value": 1}, {"value": 2}]} +def test_ascleandict_non_picklable_field(): + """Test that ascleandict handles non-picklable fields (e.g. thread locks). + + dataclasses.asdict() calls copy.deepcopy() on non-dataclass values, which + fails for objects containing thread locks with: + TypeError: cannot pickle '_thread.lock' object + The custom converter must pass such values through without copying them. + """ + import _thread + import threading + + @dataclass + class WithLock: + name: str + lock: _thread.LockType + + lock = threading.Lock() + data = WithLock(name="test", lock=lock) + result = hlp.ascleandict(data) + + with pytest.raises(TypeError): + json.dumps(result) + assert result["name"] == "test" + assert result["lock"] is lock + + result_json = hlp.ascleandict(data, json_serializable=True) + json.dumps(result_json) # Should not raise TypeError + + +def test_ascleandict_tuple_and_namedtuple(): + """Test that tuple fields are returned as plain tuples, not the original subclass. + + type(obj)(items) fails for namedtuples because their constructor requires + positional keyword arguments, not a single iterable. + """ + from collections import namedtuple + + Point = namedtuple("Point", ["x", "y"]) + + @dataclass + class WithTuples: + coords: tuple + point: tuple + name: str + + data = WithTuples(coords=(1, 2, 3), point=Point(x=4, y=5), name="test") + result = hlp.ascleandict(data) + assert result["name"] == "test" + assert result["coords"] == (1, 2, 3) + assert isinstance(result["coords"], tuple) + # namedtuple is a tuple subclass — must come back as a plain tuple + assert result["point"] == (4, 5) + assert type(result["point"]) is tuple + + +def test_ascleandict_json_serializable_non_string_keys(): + """Test that json_serializable=True converts non-string dict keys to str. + + JSON only supports string keys. A dict with e.g. integer or tuple keys + would raise TypeError on json.dumps even if all values are serializable. + """ + import uuid + from datetime import datetime + + @dataclass + class WithComplexKeys: + data: dict + + dt_key = datetime(2026, 4, 8) + uuid_key = uuid.UUID("12345678-1234-5678-1234-567812345678") + data = WithComplexKeys( + data={ + 42: "int key", + (1, 2): "tuple key", + dt_key: "datetime key", + uuid_key: "uuid key", + } + ) + + result = hlp.ascleandict(data, json_serializable=True) + # All keys must be strings + assert all(isinstance(k, str) for k in result["data"]) + assert result["data"]["42"] == "int key" + assert result["data"]["(1, 2)"] == "tuple key" + assert result["data"][str(dt_key)] == "datetime key" + assert result["data"][str(uuid_key)] == "uuid key" + json.dumps(result) # Must not raise + + def test_ascleandict_nested_cleanup_multiple_passes(): """Test that ascleandict removes cascading empty nested structures."""