opensource
/
dify
mirror of https://github.com/langgenius/dify.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refine variable truncator type hints (#27220)

This commit is contained in:
-LAN- 2025-10-22 17:08:55 +08:00 committed by GitHub
parent c61c2b0abd
commit 2d73ee64a3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 13 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
api/services/variable_truncator.py
View File

@ -57,7 +57,7 @@ class UnknownTypeError(Exception):
pass
JSONTypes: TypeAlias = int | float | str | list | dict | None | bool
JSONTypes: TypeAlias = int | float | str | list[object] | dict[str, object] | None | bool
@dataclasses.dataclass(frozen=True)
@ -252,14 +252,14 @@ class VariableTruncator:
truncated_value = value[:truncated_size] + "..."
return _PartResult(truncated_value, self.calculate_json_size(truncated_value), True)
def _truncate_array(self, value: list, target_size: int) -> _PartResult[list]:
def _truncate_array(self, value: list[object], target_size: int) -> _PartResult[list[object]]:
"""
Truncate array with correct strategy:
1. First limit to 20 items
2. If still too large, truncate individual items
"""
truncated_value: list[Any] = []
truncated_value: list[object] = []
truncated = False
used_size = self.calculate_json_size([])
@ -282,7 +282,11 @@ class VariableTruncator:
if used_size > target_size:
break
part_result = self._truncate_json_primitives(item, target_size - used_size)
remaining_budget = target_size - used_size
if item is None or isinstance(item, (str, list, dict, bool, int, float)):
part_result = self._truncate_json_primitives(item, remaining_budget)
else:
raise UnknownTypeError(f"got unknown type {type(item)} in array truncation")
truncated_value.append(part_result.value)
used_size += part_result.value_size
truncated = part_result.truncated
@ -373,10 +377,10 @@ class VariableTruncator:
def _truncate_json_primitives(self, val: str, target_size: int) -> _PartResult[str]: ...
@overload
def _truncate_json_primitives(self, val: list, target_size: int) -> _PartResult[list]: ...
def _truncate_json_primitives(self, val: list[object], target_size: int) -> _PartResult[list[object]]: ...
@overload
def _truncate_json_primitives(self, val: dict, target_size: int) -> _PartResult[dict]: ...
def _truncate_json_primitives(self, val: dict[str, object], target_size: int) -> _PartResult[dict[str, object]]: ...
@overload
def _truncate_json_primitives(self, val: bool, target_size: int) -> _PartResult[bool]: ... # type: ignore
@ -391,7 +395,9 @@ class VariableTruncator:
def _truncate_json_primitives(self, val: None, target_size: int) -> _PartResult[None]: ...
def _truncate_json_primitives(
self, val: UpdatedVariable | str | list | dict | bool | int | float | None, target_size: int
self,
val: UpdatedVariable | str | list[object] | dict[str, object] | bool | int | float | None,
target_size: int,
) -> _PartResult[Any]:
"""Truncate a value within an object to fit within budget."""
if isinstance(val, UpdatedVariable):