Skip to content
Closed
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
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "uipath"
version = "2.2.8"
version = "2.2.9"
description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
readme = { file = "README.md", content-type = "text/markdown" }
requires-python = ">=3.11"
Expand Down
45 changes: 45 additions & 0 deletions src/uipath/_utils/_bindings.py
Original file line number Diff line number Diff line change
Expand Up @@ -197,3 +197,48 @@ def get_inferred_bindings_names(cls: T):
inferred_bindings[name] = method._infer_bindings_mappings # type: ignore # probably a better way to do this

return inferred_bindings


def resolve_folder_from_bindings(
resource_type: str,
resource_name: Optional[str],
folder_path: Optional[str] = None,
) -> tuple[Optional[str], Optional[str]]:
"""Resolve folder path and key from bindings context.

This function looks up the bindings context variable to find the folder
information for a given resource.

Args:
resource_type: The type of resource (e.g., "app", "process", "index")
resource_name: The name/identifier of the resource
folder_path: Optional current folder path for more specific matching

Returns:
Tuple of (folder_path, folder_key) from bindings. Returns (None, None)
if no bindings context is available or no matching resource is found.
"""
if not resource_name:
return None, None

context_overwrites = _resource_overwrites.get()
if context_overwrites is None:
return None, None

key = f"{resource_type}.{resource_name}"
# try to apply folder path, fallback to resource_type.resource_name
if folder_path:
key = (
f"{key}.{folder_path}"
if f"{key}.{folder_path}" in context_overwrites
else key
)

matched = context_overwrites.get(key)
if matched is None:
return None, None

if hasattr(matched, "folder_path"):
return matched.folder_path, matched.folder_identifier

return None, matched.folder_identifier
29 changes: 23 additions & 6 deletions src/uipath/platform/resume_triggers/_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
)

from uipath._cli._utils._common import serialize_object
from uipath._utils._bindings import resolve_folder_from_bindings
from uipath.platform import UiPath
from uipath.platform.action_center import Task
from uipath.platform.common import (
Expand Down Expand Up @@ -260,12 +261,20 @@ async def _handle_task_trigger(
resume_trigger: The resume trigger to populate
uipath: The UiPath client instance
"""
resume_trigger.folder_path = value.app_folder_path
resume_trigger.folder_key = value.app_folder_key

if isinstance(value, (WaitTask, WaitEscalation)):
resume_trigger.item_key = value.action.key
resume_trigger.folder_path = value.app_folder_path
resume_trigger.folder_key = value.app_folder_key
elif isinstance(value, (CreateTask, CreateEscalation)):
resolved_path, resolved_key = resolve_folder_from_bindings(
resource_type="app",
resource_name=value.app_name,
folder_path=value.app_folder_path,
)

resume_trigger.folder_path = resolved_path or value.app_folder_path
resume_trigger.folder_key = resolved_key or value.app_folder_key

action = await uipath.tasks.create_async(
title=value.title,
app_name=value.app_name if value.app_name else "",
Expand All @@ -289,12 +298,20 @@ async def _handle_job_trigger(
resume_trigger: The resume trigger to populate
uipath: The UiPath client instance
"""
resume_trigger.folder_path = value.process_folder_path
resume_trigger.folder_key = value.process_folder_key

if isinstance(value, WaitJob):
resume_trigger.item_key = value.job.key
resume_trigger.folder_path = value.process_folder_path
resume_trigger.folder_key = value.process_folder_key
elif isinstance(value, InvokeProcess):
resolved_path, resolved_key = resolve_folder_from_bindings(
resource_type="process",
resource_name=value.name,
folder_path=value.process_folder_path,
)

resume_trigger.folder_path = resolved_path or value.process_folder_path
resume_trigger.folder_key = resolved_key or value.process_folder_key

job = await uipath.processes.invoke_async(
name=value.name,
input_arguments=value.input_arguments,
Expand Down
152 changes: 152 additions & 0 deletions tests/sdk/test_bindings.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,11 @@

from uipath._utils import resource_override
from uipath._utils._bindings import (
ConnectionResourceOverwrite,
GenericResourceOverwrite,
ResourceOverwritesContext,
_resource_overwrites,
resolve_folder_from_bindings,
)


Expand Down Expand Up @@ -335,3 +337,153 @@ def get_asset(name, folder_path):
# Verify context was cleaned up despite the exception
result_after = get_asset("test", "original")
assert result_after == ("test", "original")


class TestResolveFolderFromBindings:
"""Tests for the resolve_folder_from_bindings utility function."""

def test_returns_none_when_no_context(self):
"""Test that resolve_folder_from_bindings returns (None, None) when context is not set."""
result = resolve_folder_from_bindings("app", "my_app")
assert result == (None, None)

def test_returns_none_when_resource_name_is_none(self):
"""Test that resolve_folder_from_bindings returns (None, None) when resource_name is None."""
overwrites = {
"app.my_app": GenericResourceOverwrite(
resource_type="app", name="new_app", folder_path="new_folder"
)
}
token = _resource_overwrites.set(overwrites)
try:
result = resolve_folder_from_bindings("app", None)
assert result == (None, None)
finally:
_resource_overwrites.reset(token)

def test_returns_none_when_resource_name_is_empty(self):
"""Test that resolve_folder_from_bindings returns (None, None) when resource_name is empty."""
overwrites = {
"app.my_app": GenericResourceOverwrite(
resource_type="app", name="new_app", folder_path="new_folder"
)
}
token = _resource_overwrites.set(overwrites)
try:
result = resolve_folder_from_bindings("app", "")
assert result == (None, None)
finally:
_resource_overwrites.reset(token)

def test_returns_folder_path_for_generic_overwrite(self):
"""Test that resolve_folder_from_bindings returns folder_path for GenericResourceOverwrite."""
overwrites = {
"app.my_app": GenericResourceOverwrite(
resource_type="app", name="new_app", folder_path="new_folder"
)
}
token = _resource_overwrites.set(overwrites)
try:
result = resolve_folder_from_bindings("app", "my_app")
assert result == ("new_folder", "new_folder")
finally:
_resource_overwrites.reset(token)

def test_returns_folder_key_for_connection_overwrite(self):
"""Test that resolve_folder_from_bindings returns folder_key for ConnectionResourceOverwrite."""
overwrites = {
"connection.my_connection": ConnectionResourceOverwrite(
resource_type="connection",
connection_id="conn_123",
element_instance_id="elem_456",
folder_key="folder_key_789",
)
}
token = _resource_overwrites.set(overwrites)
try:
result = resolve_folder_from_bindings("connection", "my_connection")
assert result == (None, "folder_key_789")
finally:
_resource_overwrites.reset(token)

def test_returns_none_when_no_matching_key(self):
"""Test that resolve_folder_from_bindings returns (None, None) when no matching key exists."""
overwrites = {
"app.other_app": GenericResourceOverwrite(
resource_type="app", name="new_app", folder_path="new_folder"
)
}
token = _resource_overwrites.set(overwrites)
try:
result = resolve_folder_from_bindings("app", "my_app")
assert result == (None, None)
finally:
_resource_overwrites.reset(token)

def test_prefers_specific_key_with_folder_path(self):
"""Test that resolve_folder_from_bindings prefers specific key with folder_path."""
overwrites = {
"app.my_app": GenericResourceOverwrite(
resource_type="app", name="generic_app", folder_path="generic_folder"
),
"app.my_app.specific_folder": GenericResourceOverwrite(
resource_type="app", name="specific_app", folder_path="specific_folder"
),
}
token = _resource_overwrites.set(overwrites)
try:
result = resolve_folder_from_bindings("app", "my_app", "specific_folder")
assert result == ("specific_folder", "specific_folder")
finally:
_resource_overwrites.reset(token)

def test_falls_back_to_generic_key_when_specific_not_found(self):
"""Test that resolve_folder_from_bindings falls back to generic key."""
overwrites = {
"app.my_app": GenericResourceOverwrite(
resource_type="app", name="generic_app", folder_path="generic_folder"
),
}
token = _resource_overwrites.set(overwrites)
try:
result = resolve_folder_from_bindings("app", "my_app", "unknown_folder")
assert result == ("generic_folder", "generic_folder")
finally:
_resource_overwrites.reset(token)

def test_works_with_process_resource_type(self):
"""Test that resolve_folder_from_bindings works with process resource type."""
overwrites = {
"process.my_process": GenericResourceOverwrite(
resource_type="process",
name="new_process",
folder_path="process_folder",
)
}
token = _resource_overwrites.set(overwrites)
try:
result = resolve_folder_from_bindings("process", "my_process")
assert result == ("process_folder", "process_folder")
finally:
_resource_overwrites.reset(token)

@pytest.mark.anyio
async def test_works_within_resource_overwrites_context(self):
"""Test that resolve_folder_from_bindings works within ResourceOverwritesContext."""

async def get_overwrites():
return {
"app.my_app": GenericResourceOverwrite(
resource_type="app",
name="resolved_app",
folder_path="resolved_folder",
)
}

async with ResourceOverwritesContext(get_overwrites):
result = resolve_folder_from_bindings("app", "my_app")
assert result == ("resolved_folder", "resolved_folder")

# Context should be cleaned up
result_after = resolve_folder_from_bindings("app", "my_app")
assert result_after == (None, None)
2 changes: 1 addition & 1 deletion uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.