Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 34 additions & 25 deletions src/uipath_langchain/agent/react/json_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,41 +45,38 @@ def _recursive_search(
current_model: type[BaseModel], current_path: str
) -> list[str]:
"""Recursively search for fields of the target type."""
json_paths = []

target_type = _get_target_type(current_model, type_name)
matches_type = _create_type_matcher(type_name, target_type)

def _paths_for(annotation: Any, path: str) -> list[str]:
"""Paths under ``path`` whose type reaches the target type, examined
across every non-None union member and through list/nested layers.

A field can reference the target via any branch of an anyOf/oneOf
union, so every member is searched — not just the first.
"""
paths: list[str] = []
for member in _union_members(annotation):
if matches_type(member):
paths.append(path)
elif get_origin(member) is list:
inner_type, suffix = _unwrap_lists(member)
paths.extend(_paths_for(inner_type, f"{path}{suffix}"))
elif _is_pydantic_model(member):
paths.extend(_recursive_search(member, path))
return paths

json_paths: list[str] = []
for field_name, field_info in current_model.model_fields.items():
annotation = field_info.annotation

json_key = _json_key(field_name, field_info)
if current_path:
field_path = f"{current_path}.{json_key}"
else:
field_path = f"$.{json_key}"

annotation = _unwrap_optional(annotation)
origin = get_origin(annotation)

if matches_type(annotation):
json_paths.append(field_path)
continue

if origin is list:
inner_type, suffix = _unwrap_lists(annotation)
inner_path = f"{field_path}{suffix}"
if matches_type(inner_type):
json_paths.append(inner_path)
continue
if _is_pydantic_model(inner_type):
nested_paths = _recursive_search(inner_type, inner_path)
json_paths.extend(nested_paths)
continue

if _is_pydantic_model(annotation):
nested_paths = _recursive_search(annotation, field_path)
json_paths.extend(nested_paths)
for found in _paths_for(field_info.annotation, field_path):
if found not in json_paths:
json_paths.append(found)

return json_paths

Expand Down Expand Up @@ -194,6 +191,18 @@ def _unwrap_optional(annotation: Any) -> Any:
return annotation


def _union_members(annotation: Any) -> list[Any]:
"""Return the non-None members of a Union, or ``[annotation]`` otherwise.

Unlike :func:`_unwrap_optional`, which collapses a Union to a single type
(correct only for Optional-stripping), this preserves every branch so a
target type reachable through any anyOf/oneOf member is not lost.
"""
if get_origin(annotation) in (Union, types.UnionType):
return [arg for arg in get_args(annotation) if arg is not type(None)]
return [annotation]


def _unwrap_lists(annotation: Any) -> tuple[Any, str]:
"""Unwrap nested list types, returning (inner_type, jsonpath_suffix).

Expand Down
86 changes: 85 additions & 1 deletion tests/agent/react/test_json_utils.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Any, Optional
from typing import Any, Optional, Union

from pydantic import BaseModel, RootModel

Expand Down Expand Up @@ -532,3 +532,87 @@ def test_get_job_attachments_after_second_build(self) -> None:

attachments = get_job_attachments(first, self._data())
assert [str(att.id) for att in attachments] == [self._ATTACHMENT_ID]


# -- get_json_paths_by_type: multi-branch unions (anyOf / oneOf) ---------------


class _Other(BaseModel):
note: str


class _HasTarget(BaseModel):
t: Target


class TestGetJsonPathsByTypeUnions:
"""A field can reference the target type via ANY member of a multi-branch
union (a JSON-schema anyOf/oneOf), not only the first. Every member must be
searched — collapsing to the first non-None member silently drops the path
and, downstream, the value (e.g. a job attachment)."""

def test_target_in_second_union_branch(self) -> None:
class Model(BaseModel):
field: Union[_Other, Target]

assert get_json_paths_by_type(Model, "Target") == ["$.field"]

def test_target_in_optional_union(self) -> None:
class Model(BaseModel):
field: Optional[Union[_Other, Target]] = None

assert get_json_paths_by_type(Model, "Target") == ["$.field"]

def test_pep604_three_member_union(self) -> None:
class Model(BaseModel):
field: "str | Target | None" = None

assert get_json_paths_by_type(Model, "Target") == ["$.field"]

def test_list_of_union(self) -> None:
"""The list-inner type is itself a union — both layers must be opened."""

class Model(BaseModel):
items: list[Union[_Other, Target]]

assert get_json_paths_by_type(Model, "Target") == ["$.items[*]"]

def test_union_member_is_a_container(self) -> None:
class Model(BaseModel):
field: Union[_Other, _HasTarget, None]

assert get_json_paths_by_type(Model, "Target") == ["$.field.t"]

def test_multiple_branches_reference_target(self) -> None:
class Model(BaseModel):
field: Union[Target, _HasTarget]

assert get_json_paths_by_type(Model, "Target") == ["$.field", "$.field.t"]

def test_target_in_first_branch_still_works(self) -> None:
class Model(BaseModel):
field: Union[Target, _Other]

assert get_json_paths_by_type(Model, "Target") == ["$.field"]

def test_anyof_ref_in_second_branch_dynamic(self) -> None:
"""JSON-schema anyOf with the attachment $ref in a non-first branch."""
schema: dict[str, Any] = {
"type": "object",
"properties": {
"thing": {
"anyOf": [
{"type": "string"},
{"$ref": "#/definitions/job-attachment"},
]
}
},
"definitions": {
"job-attachment": {
"type": "object",
"properties": {"ID": {"type": "string"}},
}
},
}
model = create_model(schema)
assert get_json_paths_by_type(model, "__Job_attachment") == ["$.thing"]
Loading