Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

Appearance settings
Merged
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
221 changes: 218 additions & 3 deletions 221 src/uipath/_services/connections_service.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,17 @@
import json
import logging
from typing import Any, Dict
from typing import Any, Dict, List, Optional

from httpx import Response

from .._config import Config
from .._execution_context import ExecutionContext
from .._utils import Endpoint, RequestSpec, infer_bindings
from .._utils import Endpoint, RequestSpec, header_folder, infer_bindings
from ..models import Connection, ConnectionToken, EventArguments
from ..models.connections import ConnectionTokenType
from ..tracing._traced import traced
from ._base_service import BaseService
from .folder_service import FolderService

logger: logging.Logger = logging.getLogger("uipath")

Expand All @@ -20,9 +23,16 @@ class ConnectionsService(BaseService):
and secure token management.
"""

def __init__(self, config: Config, execution_context: ExecutionContext) -> None:
def __init__(
self,
config: Config,
execution_context: ExecutionContext,
folders_service: FolderService,
) -> None:
super().__init__(config=config, execution_context=execution_context)
self._folders_service = folders_service

@infer_bindings(resource_type="connection", name="key")
uipreliga marked this conversation as resolved.
Show resolved Hide resolved
@traced(
name="connections_retrieve",
run_type="uipath",
Expand All @@ -45,6 +55,117 @@ def retrieve(self, key: str) -> Connection:
response = self.request(spec.method, url=spec.endpoint)
return Connection.model_validate(response.json())

@traced(name="connections_list", run_type="uipath")
def list(
self,
*,
name: Optional[str] = None,
folder_path: Optional[str] = None,
folder_key: Optional[str] = None,
connector_key: Optional[str] = None,
skip: Optional[int] = None,
top: Optional[int] = None,
) -> List[Connection]:
"""Lists all connections with optional filtering.

Args:
name: Optional connection name to filter (supports partial matching)
folder_path: Optional folder path for filtering connections
folder_key: Optional folder key (mutually exclusive with folder_path)
connector_key: Optional connector key to filter by specific connector type
skip: Number of records to skip (for pagination)
top: Maximum number of records to return

Returns:
List[Connection]: List of connection instances

Raises:
ValueError: If both folder_path and folder_key are provided together, or if
folder_path is provided but cannot be resolved to a folder_key

Examples:
>>> # List all connections
>>> connections = sdk.connections.list()

>>> # Find connections by name
>>> salesforce_conns = sdk.connections.list(name="Salesforce")

>>> # List all Slack connections in Finance folder
>>> connections = sdk.connections.list(
... folder_path="Finance",
... connector_key="uipath-slack"
... )
"""
spec = self._list_spec(
name=name,
folder_path=folder_path,
folder_key=folder_key,
connector_key=connector_key,
skip=skip,
top=top,
)
response = self.request(
spec.method, url=spec.endpoint, params=spec.params, headers=spec.headers
)

return self._parse_and_validate_list_response(response)

@traced(name="connections_list", run_type="uipath")
async def list_async(
self,
*,
name: Optional[str] = None,
folder_path: Optional[str] = None,
folder_key: Optional[str] = None,
connector_key: Optional[str] = None,
skip: Optional[int] = None,
top: Optional[int] = None,
) -> List[Connection]:
"""Asynchronously lists all connections with optional filtering.

Args:
name: Optional connection name to filter (supports partial matching)
folder_path: Optional folder path for filtering connections
folder_key: Optional folder key (mutually exclusive with folder_path)
connector_key: Optional connector key to filter by specific connector type
skip: Number of records to skip (for pagination)
top: Maximum number of records to return

Returns:
List[Connection]: List of connection instances

Raises:
ValueError: If both folder_path and folder_key are provided together, or if
folder_path is provided but cannot be resolved to a folder_key

Examples:
>>> # List all connections
>>> connections = await sdk.connections.list_async()

>>> # Find connections by name
>>> salesforce_conns = await sdk.connections.list_async(name="Salesforce")

>>> # List all Slack connections in Finance folder
>>> connections = await sdk.connections.list_async(
... folder_path="Finance",
... connector_key="uipath-slack"
... )
"""
spec = self._list_spec(
name=name,
folder_path=folder_path,
folder_key=folder_key,
connector_key=connector_key,
skip=skip,
top=top,
)
response = await self.request_async(
spec.method, url=spec.endpoint, params=spec.params, headers=spec.headers
)

return self._parse_and_validate_list_response(response)

@infer_bindings(resource_type="connection", name="key")
@traced(
name="connections_retrieve",
run_type="uipath",
Expand Down Expand Up @@ -213,3 +334,97 @@ def _retrieve_token_spec(
endpoint=Endpoint(f"/connections_/api/v1/Connections/{key}/token"),
params={"tokenType": token_type.value},
)

def _parse_and_validate_list_response(self, response: Response) -> List[Connection]:
"""Parse and validate the list response from the API.

Handles both OData response format (with 'value' field) and raw list responses.

Args:
response: The HTTP response from the API

Returns:
List of validated Connection instances
"""
data = response.json()

# Handle both OData responses (dict with 'value') and raw list responses
if isinstance(data, dict):
connections_data = data.get("value", [])
elif isinstance(data, list):
connections_data = data
else:
connections_data = []

return [Connection.model_validate(conn) for conn in connections_data]

def _list_spec(
self,
name: Optional[str] = None,
folder_path: Optional[str] = None,
folder_key: Optional[str] = None,
connector_key: Optional[str] = None,
skip: Optional[int] = None,
top: Optional[int] = None,
) -> RequestSpec:
"""Build the request specification for listing connections.

Args:
name: Optional connection name to filter (supports partial matching)
folder_path: Optional folder path for filtering connections
folder_key: Optional folder key (mutually exclusive with folder_path)
connector_key: Optional connector key to filter by specific connector type
skip: Number of records to skip (for pagination)
top: Maximum number of records to return

Returns:
RequestSpec with endpoint, params, and headers configured

Raises:
ValueError: If both folder_path and folder_key are provided together, or if
folder_path is provided but cannot be resolved to a folder_key
"""
uipreliga marked this conversation as resolved.
Show resolved Hide resolved
# Validate mutual exclusivity of folder_path and folder_key
if folder_path is not None and folder_key is not None:
raise ValueError(
"folder_path and folder_key are mutually exclusive and cannot be provided together"
)

# Resolve folder_path to folder_key if needed
resolved_folder_key = folder_key
if not resolved_folder_key and folder_path:
resolved_folder_key = self._folders_service.retrieve_key(
folder_path=folder_path
)
if not resolved_folder_key:
raise ValueError(f"Folder with path '{folder_path}' not found")

# Build OData filters
filters = []
if name:
# Escape single quotes in name for OData
escaped_name = name.replace("'", "''")
filters.append(f"contains(Name, '{escaped_name}')")
if connector_key:
filters.append(f"connector/key eq '{connector_key}'")

params = {}
if filters:
params["$filter"] = " and ".join(filters)
if skip is not None:
params["$skip"] = str(skip)
if top is not None:
params["$top"] = str(top)

# Always expand connector and folder for complete information
params["$expand"] = "connector,folder"

# Use header_folder which handles validation
headers = header_folder(resolved_folder_key, None)

return RequestSpec(
method="GET",
endpoint=Endpoint("/connections_/api/v1/Connections"),
params=params,
headers=headers,
)
11 changes: 10 additions & 1 deletion 11 src/uipath/_uipath.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ def __init__(
self._folders_service: Optional[FolderService] = None
self._buckets_service: Optional[BucketsService] = None
self._attachments_service: Optional[AttachmentsService] = None
self._connections_service: Optional[ConnectionsService] = None

setup_logging(debug)
self._execution_context = ExecutionContext()
Expand Down Expand Up @@ -98,7 +99,15 @@ def buckets(self) -> BucketsService:

@property
def connections(self) -> ConnectionsService:
return ConnectionsService(self._config, self._execution_context)
if not self._connections_service:
if not self._folders_service:
self._folders_service = FolderService(
self._config, self._execution_context
)
self._connections_service = ConnectionsService(
self._config, self._execution_context, self._folders_service
)
return self._connections_service

@property
def context_grounding(self) -> ContextGroundingService:
Expand Down
6 changes: 4 additions & 2 deletions 6 src/uipath/_utils/_infer_bindings.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,10 @@ def wrapper(*args, **kwargs):
all_args.get(name), # type: ignore
all_args.get(folder_path, None),
) as (name_overwrite_or_default, folder_path_overwrite_or_default):
all_args[name] = name_overwrite_or_default
all_args[folder_path] = folder_path_overwrite_or_default
if name in sig.parameters:
all_args[name] = name_overwrite_or_default
if folder_path in sig.parameters:
all_args[folder_path] = folder_path_overwrite_or_default

return func(**all_args)

Expand Down
Loading
Morty Proxy This is a proxified and sanitized view of the page, visit original site.