Skip to content
Open
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
3 changes: 2 additions & 1 deletion src/uipath/_cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
from .cli_pull import pull as pull # type: ignore
from .cli_push import push as push # type: ignore
from .cli_run import run as run # type: ignore

from .cli_list import get as get # type: ignore

def _get_safe_version() -> str:
"""Get the version of the uipath package."""
Expand Down Expand Up @@ -73,3 +73,4 @@ def cli(lv: bool, v: bool) -> None:
cli.add_command(pull)
cli.add_command(eval)
cli.add_command(dev)
cli.add_command(get)
63 changes: 63 additions & 0 deletions src/uipath/_cli/cli_list.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
import json
from typing import List, Protocol, Any, Dict, Callable

import click

from ._utils._common import environment_options, get_env_vars
from ._utils._console import ConsoleLogger
from ..telemetry import track

from .._config import Config
from .._execution_context import ExecutionContext
from .._services import EntitiesService, AssetsService

console = ConsoleLogger()


class CLIListable(Protocol):
"""Protocol for services that support listing"""
def _list(self, **kwargs) -> List[Any]:
...

def create_service(resource_type: str) -> CLIListable:
"""Service factory - creates the appropriate service"""
[base_url, token] = get_env_vars()

config = Config(base_url=base_url, secret=token)
execution_context = ExecutionContext()

services: Dict[str, Callable[[], CLIListable]] = {
'entities': lambda: EntitiesService(config, execution_context),
'assets': lambda: AssetsService(config, execution_context),
}

if resource_type not in services:
available = ', '.join(services.keys())
raise ValueError(f"Unknown resource type '{resource_type}'. Available: {available}")

return services[resource_type]()

@click.command()
@click.argument('resource_type')
@click.option('--folder-path', '-f', help='Folder path to list resources from')
def get(
resource_type: str,
folder_path: str = None):
"""List resources of a given type."""

with console.spinner(f"Fetching {resource_type}..."):
try:
service: CLIListable = create_service(resource_type)
# Only pass folder_path if it's provided
kwargs = {}
if folder_path:
kwargs['folder_path'] = folder_path
list_results = service._list(**kwargs)

data = [item.model_dump(mode='json') for item in list_results]
console.success(json.dumps(data, indent=2))

except Exception as e:
console.error(f"Error fetching {resource_type}: {e}")
raise click.Abort()

10 changes: 10 additions & 0 deletions src/uipath/_services/_cli_service.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
from abc import ABC, abstractmethod


class ClIService(ABC):
"""
Abstract class implemented by services supported in the CLI commands.
"""
@abstractmethod
def _list(self, **kwargs):
pass
69 changes: 67 additions & 2 deletions src/uipath/_services/assets_service.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Dict, Optional
from typing import Dict, Optional, List

from httpx import Response

Expand All @@ -10,9 +10,10 @@
from ..models import Asset, UserAsset
from ..tracing._traced import traced
from ._base_service import BaseService
from ._cli_service import ClIService


class AssetsService(FolderContext, BaseService):
class AssetsService(FolderContext, BaseService, ClIService):
"""Service for managing UiPath assets.

Assets are key-value pairs that can be used to store configuration data,
Expand All @@ -27,6 +28,56 @@ def __init__(self, config: Config, execution_context: ExecutionContext) -> None:
@traced(
name="assets_retrieve", run_type="uipath", hide_input=True, hide_output=True
)

def _list(self, **kwargs) -> List[Asset]:
"""List all assets in the current folder"""

folder_path = kwargs.get('folder_path')
if folder_path is None:
if self._folder_path is None:
raise ValueError("Either folder_path must be provided or folder_key must be configured")
spec = self._list_spec(folder_key=self._folder_key)
else:
spec = self._list_spec(folder_path=folder_path)

response = self.request(
spec.method,
url=spec.endpoint,
params=spec.params,
content=spec.content,
headers=spec.headers,
)

assets_data = response.json().get("value", [])

return [Asset.model_validate(asset_data) for asset_data in assets_data]

@infer_bindings(resource_type="asset")
async def _list_async(
self,
folder_path: Optional[str] = None,
name: Optional[str] = None) -> List[Asset]:
"""Asynchronously list all assets in the current folder"""

if folder_path is None:
if self._folder_path is None:
raise ValueError("Either folder_path must be provided or folder_key must be configured")
spec = self._list_spec(folder_key=self._folder_key)
else:
spec = self._list_spec(folder_path=folder_path)

response = await self.request_async(
spec.method,
url=spec.endpoint,
params=spec.params,
content=spec.content,
headers=spec.headers,
)

assets_data = response.json().get("value", [])

return [Asset.model_validate(asset_data) for asset_data in assets_data]

@infer_bindings(resource_type="asset")
def retrieve(
self,
Expand Down Expand Up @@ -371,3 +422,17 @@ def _update_spec(
**header_folder(folder_key, folder_path),
},
)

def _list_spec(
self,
folder_key: Optional[str] = None,
folder_path: Optional[str] = None,
) -> RequestSpec:
return RequestSpec(
method="GET",
endpoint=Endpoint("/orchestrator_/odata/Assets/UiPath.Server.Configuration.OData.GetFiltered"),
params={"$top": 100},
headers={
**header_folder(folder_key, folder_path),
},
)
6 changes: 5 additions & 1 deletion src/uipath/_services/entities_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,10 @@
)
from ..tracing import traced
from ._base_service import BaseService
from ._cli_service import ClIService


class EntitiesService(BaseService):
class EntitiesService(BaseService, ClIService):
"""Service for managing UiPath Data Service entities.

Entities are database tables in UiPath Data Service that can store
Expand All @@ -24,6 +25,9 @@ class EntitiesService(BaseService):
def __init__(self, config: Config, execution_context: ExecutionContext) -> None:
super().__init__(config=config, execution_context=execution_context)

def _list(self, **kwargs) -> List[Entity]:
return self.list_entities()

@traced(name="entity_retrieve", run_type="uipath")
def retrieve(self, entity_key: str) -> Entity:
"""Retrieve an entity by its key.
Expand Down
Loading