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
15 changes: 15 additions & 0 deletions src/uipath/_cli/_contracts/_contracts.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
from enum import Enum
from pydantic import BaseModel

class Severity(str, Enum):
"""Severity level for virtual resource operation results."""

INFO = "info"
SUCCESS = "success"
ATTENTION = "attention"
ERROR = "error"
WARN = "warn"

class UiPathUpdateEvent(BaseModel):
message: str
severity: Severity
101 changes: 99 additions & 2 deletions src/uipath/_cli/_utils/_common.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,23 @@
import json
import os
from enum import Enum

from pathlib import Path
from typing import Optional
from typing import Optional, Generator
from urllib.parse import urlparse

import click
from dotenv import load_dotenv

from pydantic import BaseModel

from .._contracts._contracts import UiPathUpdateEvent, Severity
from ..models.runtime_schema import Bindings
from ..._config import UiPathConfig
from ..._utils._bindings import ResourceOverwrite, ResourceOverwriteParser
from ..._utils.constants import DOTENV_FILE
from ..spinner import Spinner


def add_cwd_to_path():
import sys

Expand Down Expand Up @@ -100,6 +105,98 @@ def get_org_scoped_url(base_url: str) -> str:
org_scoped_url = f"{parsed.scheme}://{parsed.netloc}/{org_name}"
return org_scoped_url

def create_binding_files() -> Generator[UiPathUpdateEvent, None, None]:
import importlib.resources
import shutil

bindings_file_path = UiPathConfig.bindings_file_path

if bindings_file_path.exists():
yield UiPathUpdateEvent(
message="Bindings file already exists. Skipping...",
severity=Severity.ATTENTION
)
else:
default_bindings = Bindings(
version="2.0",
resources=[],
)

try:
with open(bindings_file_path, "w") as f:
json.dump(default_bindings.model_dump(by_alias=True), f, indent=2)

yield UiPathUpdateEvent(
message="Created 'bindings.json' file.",
severity=Severity.SUCCESS
)

except Exception as e:
yield UiPathUpdateEvent(
message=f"Failed to create 'bindings.json' file. Error: {e}",
severity=Severity.ERROR
)
return

# Ensure .agent directory exists
agent_dir = Path(os.getcwd()) / ".agent"
agent_dir.mkdir(exist_ok=True)

# Handle BINDINGS.md documentation file in .agent directory
bindings_md_path = agent_dir / "BINDINGS.md"

if bindings_md_path.exists():
yield UiPathUpdateEvent(
message="Bindings documentation (.agent/BINDINGS.md) already exists. Skipping...",
severity=Severity.INFO
)
else:
try:
# Copy BINDINGS.md from resources
source_path = importlib.resources.files("uipath._resources").joinpath("BINDINGS.md")

with importlib.resources.as_file(source_path) as src_path:
shutil.copy(src_path, bindings_md_path)

yield UiPathUpdateEvent(
message="Created '.agent/BINDINGS.md' documentation file.",
severity=Severity.SUCCESS
)

except Exception as e:
yield UiPathUpdateEvent(
message=f"Failed to create bindings documentation: {e}",
severity=Severity.WARN
)

# Handle bindings.json.example file
bindings_example_path = Path(os.getcwd()) / "bindings.json.example"

if bindings_example_path.exists():
yield UiPathUpdateEvent(
message="Bindings example file already exists. Skipping...",
severity=Severity.INFO
)
else:
try:
# Copy bindings.json.example from resources
source_path = importlib.resources.files("uipath._resources").joinpath("bindings.json.example")

with importlib.resources.as_file(source_path) as src_path:
shutil.copy(src_path, bindings_example_path)

yield UiPathUpdateEvent(
message="Created 'bindings.json.example' file.",
severity=Severity.SUCCESS
)

except Exception as e:
yield UiPathUpdateEvent(
message=f"Failed to create bindings example file: {e}",
severity=Severity.WARN
)



def clean_directory(directory: str) -> None:
"""Clean up Python files in the specified directory.
Expand Down
103 changes: 102 additions & 1 deletion src/uipath/_cli/_utils/_console.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
from contextlib import contextmanager
from enum import Enum
from typing import Any, Dict, Iterator, List, Optional, Type, TypeVar
from typing import Any, Dict, Iterator, List, Optional, Type, TypeVar, Union, AsyncIterator, Callable, Generator, \
AsyncGenerator
import asyncio
import inspect

import click
from rich.console import Console
Expand All @@ -15,12 +18,15 @@
from rich.spinner import Spinner as RichSpinner
from rich.text import Text

from uipath._cli._contracts._contracts import UiPathUpdateEvent, Severity


class LogLevel(Enum):
"""Enum for log levels with corresponding emojis."""

INFO = ""
SUCCESS = click.style("✓ ", fg="green", bold=True)
ATTENTION = "🔵"
WARNING = "⚠️"
ERROR = "❌"
HINT = "💡"
Expand Down Expand Up @@ -130,6 +136,10 @@ def hint(self, message: str) -> None:
"""Log a hint message."""
self.log(message, LogLevel.HINT)

def attention(self, message: str) -> None:
"""Log a hint message."""
self.log(message, LogLevel.ATTENTION)

def magic(self, message: str) -> None:
"""Log a magic message."""
self.log(message, LogLevel.MAGIC, "green")
Expand All @@ -155,6 +165,97 @@ def link(self, message: str, url: str) -> None:
LogLevel.LINK,
)

def process_events(
self,
event_source: Union[Callable[..., Generator[UiPathUpdateEvent, None, None]], Callable[..., AsyncGenerator[UiPathUpdateEvent, None]]],
*args,
**kwargs
) -> Any:
"""Process UpdateEvent objects from a generator or async generator.

This method handles both sync and async generators that yield UpdateEvent objects,
automatically routing each event to the appropriate console method based on severity.

Args:
event_source: A callable that returns either an Iterator or AsyncIterator of UpdateEvent objects
*args: Arguments to pass to the event_source callable
**kwargs: Keyword arguments to pass to the event_source callable

Returns:
The return value from the generator (if any)

Example:
# For a sync generator function:
def my_operation() -> Iterator[UpdateEvent]:
yield UpdateEvent(message="Starting...", severity=Severity.INFO)
# do work...
yield UpdateEvent(message="Done!", severity=Severity.SUCCESS)

# Use it like:
console.process_events(my_operation)

# For an async generator:
async def async_operation() -> AsyncIterator[UpdateEvent]:
yield UpdateEvent(message="Processing...", severity=Severity.INFO)

# Use it like:
console.process_events(async_operation)
"""
generator = event_source(*args, **kwargs)

if inspect.isasyncgen(generator):
# Handle async generator
return asyncio.run(self._process_async_events(generator))
else:
# Handle sync generator
return self._process_sync_events(generator)

def _process_sync_events(self, generator: Generator[UiPathUpdateEvent, None, None]) -> Any:
"""Process events from a synchronous generator.

Args:
generator: Iterator yielding UpdateEvent objects

Returns:
The return value from the generator (if any)
"""
result = None
for event in generator:
self._log_event(event)


return result

async def _process_async_events(self, generator: AsyncGenerator[UiPathUpdateEvent, None]) -> Any:
"""Process events from an asynchronous generator.

Args:
generator: AsyncIterator yielding UpdateEvent objects

Returns:
The return value from the generator (if any)
"""
result = None
async for event in generator:
self._log_event(event)

return result

def _log_event(self, event: UiPathUpdateEvent) -> None:
match event.severity:
case Severity.ERROR:
self.error(event.message, include_traceback=False)
case Severity.WARN:
self.warning(event.message)
case Severity.INFO:
self.info(event.message)
case Severity.ATTENTION:
self.attention(event.message)
case Severity.SUCCESS:
self.success(event.message)
case _:
self.info(event.message)

def prompt(self, message: str, **kwargs: Any) -> Any:
"""Wrapper for click.prompt with emoji.

Expand Down
8 changes: 4 additions & 4 deletions src/uipath/_cli/cli_init.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

import click

from ._utils._common import create_binding_files, Severity
from .._config import UiPathConfig
from .._utils.constants import ENV_TELEMETRY_ENABLED
from ..telemetry import track
Expand Down Expand Up @@ -223,15 +224,14 @@ async def initialize() -> None:
try:
runtime = generate_runtime_factory().new_runtime(**context_args)

bindings = await runtime.get_bindings()
bindings_path = write_bindings_file(bindings)

config_data = RuntimeSchema(
entryPoints=[await runtime.get_entrypoint()],
)
config_path = write_config_file(config_data)

console.process_events(create_binding_files)

console.success(f"Created '{config_path}' file.")
console.success(f"Created '{bindings_path}' file.")
except Exception as e:
console.error(f"Error creating configuration file:\n {str(e)}")

Expand Down
8 changes: 0 additions & 8 deletions src/uipath/_cli/cli_pack.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,14 +102,6 @@ def generate_entrypoints_file(entryPoints):

return entrypoint_json_data


def generate_bindings_content() -> Bindings:
return Bindings(
version="2.0",
resources=[],
)


def generate_content_types_content():
templates_path = os.path.join(
os.path.dirname(__file__), "_templates", "[Content_Types].xml.template"
Expand Down
4 changes: 4 additions & 0 deletions src/uipath/_resources/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -19,3 +19,7 @@ This documentation is split into multiple files for efficient context loading. L
3. **@.agent/CLI_REFERENCE.md** - CLI commands documentation
- **When to load:** Working with `uipath init`, `uipath run`, or `uipath eval` commands
- **Contains:** Command syntax, options, usage examples, and workflows

4. **@.agent/BINDINGS.md** - Resource bindings configuration guide
- **When to load:** Configuring runtime resource replacement for Orchestrator/StudioWeb deployments
- **Contains:** Binding structure, resource types, matching rules, and environment strategies
Loading
Loading