diff --git a/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py b/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py index eb3664262..06b6465e7 100644 --- a/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py +++ b/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py @@ -49,17 +49,34 @@ def format_relative_time(iso_timestamp: str) -> str: return "" +def _result_to_dict(result: Any) -> Dict[str, Any]: + """Normalize a single search result into a plain dict. + + The Supermemory SDK returns search results as Pydantic model objects, not + dicts, so calling ``.get(...)`` on them raises AttributeError. Convert + models via ``model_dump`` (keeping API field names like ``updatedAt``) and + pass existing dicts through unchanged. + """ + if isinstance(result, dict): + return result + model_dump = getattr(result, "model_dump", None) + if callable(model_dump): + return model_dump(by_alias=True) + return {} + + def deduplicate_memories( static: List[str], dynamic: List[str], - search_results: List[Dict[str, Any]], + search_results: List[Any], ) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]: """Deduplicate memories. Priority: static > dynamic > search. Args: static: List of static memory strings. dynamic: List of dynamic memory strings. - search_results: List of search result dicts with 'memory' and 'updatedAt'. + search_results: List of search results (SDK model objects or dicts) + carrying 'memory' and 'updatedAt'. """ seen = set() @@ -71,9 +88,10 @@ def unique_strings(memories: List[str]) -> List[str]: out.append(m) return out - def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + def unique_search(results: List[Any]) -> List[Dict[str, Any]]: out = [] for r in results: + r = _result_to_dict(r) memory = r.get("memory", "") if memory and memory not in seen: seen.add(memory) diff --git a/packages/cartesia-sdk-python/tests/test_search_result_models.py b/packages/cartesia-sdk-python/tests/test_search_result_models.py new file mode 100644 index 000000000..9ed1da88e --- /dev/null +++ b/packages/cartesia-sdk-python/tests/test_search_result_models.py @@ -0,0 +1,55 @@ +"""Regression tests: profile search results arrive as SDK model objects. + +The Supermemory SDK returns ``response.search_results.results`` as Pydantic +model objects, not dicts. Calling ``.get("memory")`` on a model raises +AttributeError, which crashed ``deduplicate_memories`` on every non-empty +search (the default ``mode="full"`` path). These tests reproduce that path +with a dict-less model stand-in. +""" + +import unittest + +from supermemory_cartesia.utils import deduplicate_memories, format_memories_to_text + + +class _FakeSearchResult: + """Mimics a Supermemory SDK search result: attribute access and + ``model_dump()`` but deliberately no dict ``.get()``.""" + + def __init__(self, memory, updated_at=None): + self.memory = memory + self.updatedAt = updated_at + + def model_dump(self, by_alias=False): + data = {"memory": self.memory} + if self.updatedAt is not None: + data["updatedAt"] = self.updatedAt + return data + + +class TestSearchResultModels(unittest.TestCase): + def test_model_results_do_not_crash_and_dedupe(self): + results = [ + _FakeSearchResult("likes python", "2020-01-01T00:00:00Z"), + _FakeSearchResult("likes python"), # duplicate, dropped + _FakeSearchResult("prefers async"), + ] + dedup = deduplicate_memories(static=[], dynamic=[], search_results=results) + self.assertEqual( + [r["memory"] for r in dedup["search_results"]], + ["likes python", "prefers async"], + ) + + def test_model_results_render_to_text(self): + results = [_FakeSearchResult("likes python", "2020-01-01T00:00:00Z")] + dedup = deduplicate_memories(static=[], dynamic=[], search_results=results) + self.assertIn("likes python", format_memories_to_text(dedup)) + + def test_dict_results_still_supported(self): + results = [{"memory": "from dict"}] + dedup = deduplicate_memories(static=[], dynamic=[], search_results=results) + self.assertEqual([r["memory"] for r in dedup["search_results"]], ["from dict"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py index a27da2561..f9db7354b 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py @@ -49,17 +49,34 @@ def format_relative_time(iso_timestamp: str) -> str: return "" +def _result_to_dict(result: Any) -> Dict[str, Any]: + """Normalize a single search result into a plain dict. + + The Supermemory SDK returns search results as Pydantic model objects, not + dicts, so calling ``.get(...)`` on them raises AttributeError. Convert + models via ``model_dump`` (keeping API field names like ``updatedAt``) and + pass existing dicts through unchanged. + """ + if isinstance(result, dict): + return result + model_dump = getattr(result, "model_dump", None) + if callable(model_dump): + return model_dump(by_alias=True) + return {} + + def deduplicate_memories( static: List[str], dynamic: List[str], - search_results: List[Dict[str, Any]], + search_results: List[Any], ) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]: """Deduplicate memories. Priority: static > dynamic > search. Args: static: List of static memory strings. dynamic: List of dynamic memory strings. - search_results: List of search result dicts with 'memory' and 'updatedAt'. + search_results: List of search results (SDK model objects or dicts) + carrying 'memory' and 'updatedAt'. """ seen = set() @@ -71,9 +88,10 @@ def unique_strings(memories: List[str]) -> List[str]: out.append(m) return out - def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + def unique_search(results: List[Any]) -> List[Dict[str, Any]]: out = [] for r in results: + r = _result_to_dict(r) memory = r.get("memory", "") if memory and memory not in seen: seen.add(memory) diff --git a/packages/pipecat-sdk-python/tests/test_search_result_models.py b/packages/pipecat-sdk-python/tests/test_search_result_models.py new file mode 100644 index 000000000..106fafb66 --- /dev/null +++ b/packages/pipecat-sdk-python/tests/test_search_result_models.py @@ -0,0 +1,55 @@ +"""Regression tests: profile search results arrive as SDK model objects. + +The Supermemory SDK returns ``response.search_results.results`` as Pydantic +model objects, not dicts. Calling ``.get("memory")`` on a model raises +AttributeError, which crashed ``deduplicate_memories`` on every non-empty +search (the default ``mode="full"`` path). These tests reproduce that path +with a dict-less model stand-in. +""" + +import unittest + +from supermemory_pipecat.utils import deduplicate_memories, format_memories_to_text + + +class _FakeSearchResult: + """Mimics a Supermemory SDK search result: attribute access and + ``model_dump()`` but deliberately no dict ``.get()``.""" + + def __init__(self, memory, updated_at=None): + self.memory = memory + self.updatedAt = updated_at + + def model_dump(self, by_alias=False): + data = {"memory": self.memory} + if self.updatedAt is not None: + data["updatedAt"] = self.updatedAt + return data + + +class TestSearchResultModels(unittest.TestCase): + def test_model_results_do_not_crash_and_dedupe(self): + results = [ + _FakeSearchResult("likes python", "2020-01-01T00:00:00Z"), + _FakeSearchResult("likes python"), # duplicate, dropped + _FakeSearchResult("prefers async"), + ] + dedup = deduplicate_memories(static=[], dynamic=[], search_results=results) + self.assertEqual( + [r["memory"] for r in dedup["search_results"]], + ["likes python", "prefers async"], + ) + + def test_model_results_render_to_text(self): + results = [_FakeSearchResult("likes python", "2020-01-01T00:00:00Z")] + dedup = deduplicate_memories(static=[], dynamic=[], search_results=results) + self.assertIn("likes python", format_memories_to_text(dedup)) + + def test_dict_results_still_supported(self): + results = [{"memory": "from dict"}] + dedup = deduplicate_memories(static=[], dynamic=[], search_results=results) + self.assertEqual([r["memory"] for r in dedup["search_results"]], ["from dict"]) + + +if __name__ == "__main__": + unittest.main()