forked from UiPath/uipath-dev-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path__init__.py
More file actions
298 lines (247 loc) · 10.4 KB
/
Copy path__init__.py
File metadata and controls
298 lines (247 loc) · 10.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
"""UiPath Developer Server - Web/API mode for the developer console."""
from __future__ import annotations
import asyncio
import logging
import os
import socket
import sys
import threading
import time
import webbrowser
from collections.abc import Callable
from typing import Any
import uvicorn
from uipath.core.tracing import UiPathTraceManager
from uipath.runtime import UiPathRuntimeFactoryProtocol
from uipath.dev.models.data import (
ChatData,
InterruptData,
LogData,
StateData,
TraceData,
)
from uipath.dev.models.execution import ExecutionRun
from uipath.dev.server.debug_bridge import WebDebugBridge
from uipath.dev.services.run_service import RunService
logger = logging.getLogger(__name__)
class UiPathDeveloperServer:
"""Web server mode for the UiPath Developer Console.
Provides the same functionality as UiPathDeveloperConsole but via
a FastAPI + WebSocket backend instead of a Textual TUI.
Usage::
server = UiPathDeveloperServer(
runtime_factory=factory,
trace_manager=trace_manager,
)
await server.run_async() # builds frontend, starts uvicorn, opens browser
# or
app = server.create_app() # just get the FastAPI app
"""
def __init__(
self,
runtime_factory: UiPathRuntimeFactoryProtocol,
trace_manager: UiPathTraceManager,
host: str = os.environ.get("UIPATH_DEV_SERVER_HOST", "localhost"),
port: int = int(os.environ.get("UIPATH_DEV_SERVER_PORT", "8080")),
open_browser: bool = True,
factory_creator: Callable[[], UiPathRuntimeFactoryProtocol] | None = None,
) -> None:
"""Initialize the developer server."""
self.runtime_factory = runtime_factory
self.trace_manager = trace_manager
self.host = host
self.port = port
self.open_browser = open_browser
self.factory_creator = factory_creator
self._watcher_task: asyncio.Task[None] | None = None
self._watcher_stop: asyncio.Event | None = None
self.reload_pending = False
from uipath.dev.server.ws.manager import ConnectionManager
self.connection_manager = ConnectionManager()
self.run_service = RunService(
runtime_factory=self.runtime_factory,
trace_manager=self.trace_manager,
on_run_updated=self._on_run_updated,
on_log=self._on_log,
on_trace=self._on_trace,
on_chat=self._on_chat,
on_state=self._on_state,
on_interrupt=self._on_interrupt,
debug_bridge_factory=lambda mode: WebDebugBridge(mode=mode),
on_run_removed=self.connection_manager.remove_run_subscriptions,
)
def create_app(self) -> Any:
"""Create and return a FastAPI application."""
from uipath.dev.server.app import create_app
return create_app(self)
async def run_async(self) -> None:
"""Build frontend, start the server, and open the browser.
This is the main entry point — mirrors UiPathDeveloperConsole.run_async().
Blocks until the server is shut down (Ctrl-C / SIGINT).
"""
await self.run_service.apply_factory_settings()
self.port = self._find_free_port(self.host, self.port)
app = self.create_app()
base_url = f"http://{self.host}:{self.port}"
self._print_banner(base_url)
if self.open_browser:
threading.Thread(
target=self._deferred_open_browser,
daemon=True,
).start()
# Start file watcher if factory_creator is available
if self.factory_creator is not None:
self._start_watcher()
config = uvicorn.Config(
app,
host=self.host,
port=self.port,
log_level="warning",
)
server = uvicorn.Server(config)
await server.serve()
async def shutdown(self) -> None:
"""Clean up resources before shutting down."""
logger.info("Shutting down server resources...")
self._stop_watcher()
# Close any active WebSocket connections
await self.connection_manager.disconnect_all()
# Give threads time to finish
await asyncio.sleep(0.1)
def run(self) -> None:
"""Synchronous wrapper around :meth:`run_async`."""
try:
asyncio.run(self.run_async())
except KeyboardInterrupt:
pass
# ------------------------------------------------------------------
# Hot-reload support
# ------------------------------------------------------------------
async def reload_factory(self) -> None:
"""Dispose old factory, flush user modules, and recreate."""
if self.factory_creator is None:
return
# Dispose old factory if it supports it
if hasattr(self.runtime_factory, "dispose"):
try:
await self.runtime_factory.dispose()
except Exception:
logger.debug("Error disposing old factory", exc_info=True)
# Flush user modules (files under cwd, excluding venvs/site-packages)
cwd = os.getcwd()
to_remove = [
name
for name, mod in sys.modules.items()
if hasattr(mod, "__file__")
and mod.__file__ is not None
and os.path.abspath(mod.__file__).startswith(cwd)
and ".venv" not in mod.__file__
and "site-packages" not in mod.__file__
]
for name in to_remove:
del sys.modules[name]
logger.debug("Flushed %d user modules", len(to_remove))
# Recreate factory
self.runtime_factory = self.factory_creator()
self.run_service.runtime_factory = self.runtime_factory
await self.run_service.apply_factory_settings()
self.reload_pending = False
logger.debug("Factory reloaded successfully")
def _start_watcher(self) -> None:
"""Start the file watcher background task."""
from uipath.dev.server.watcher import watch_python_files
self._watcher_stop = asyncio.Event()
self._watcher_task = asyncio.create_task(
watch_python_files(
on_change=self._on_files_changed,
stop_event=self._watcher_stop,
)
)
def _stop_watcher(self) -> None:
"""Stop the file watcher background task."""
if self._watcher_stop is not None:
self._watcher_stop.set()
if self._watcher_task is not None:
self._watcher_task.cancel()
self._watcher_task = None
def _on_files_changed(self, changed_files: list[str]) -> None:
"""Handle file change events from the watcher."""
self.reload_pending = True
self.connection_manager.broadcast_reload(changed_files)
# ------------------------------------------------------------------
# Internal callbacks
# ------------------------------------------------------------------
def _on_run_updated(self, run: ExecutionRun) -> None:
"""Broadcast run update to subscribed WebSocket clients."""
self.connection_manager.broadcast_run_updated(run)
def _on_log(self, log_data: LogData) -> None:
"""Broadcast log to subscribed WebSocket clients."""
self.connection_manager.broadcast_log(log_data)
def _on_trace(self, trace_data: TraceData) -> None:
"""Broadcast trace to subscribed WebSocket clients."""
self.connection_manager.broadcast_trace(trace_data)
def _on_chat(self, chat_data: ChatData) -> None:
"""Broadcast chat message to subscribed WebSocket clients."""
self.connection_manager.broadcast_chat(chat_data)
def _on_interrupt(self, interrupt_data: InterruptData) -> None:
"""Broadcast chat interrupt to subscribed WebSocket clients."""
self.connection_manager.broadcast_interrupt(interrupt_data)
def _on_state(self, state_data: StateData) -> None:
"""Broadcast state transition to subscribed WebSocket clients."""
self.connection_manager.broadcast_state(state_data)
@staticmethod
def _find_free_port(host: str, start_port: int, max_attempts: int = 100) -> int:
"""Find a free port starting from *start_port*.
Tries *start_port*, then *start_port + 1*, etc. up to *max_attempts*.
"""
for offset in range(max_attempts):
port = start_port + offset
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind((host, port))
return port
except OSError:
continue
raise OSError(
f"Could not find a free port in range {start_port}-{start_port + max_attempts - 1}"
)
@staticmethod
def _print_banner(base_url: str) -> None:
"""Print a welcome banner to the console."""
import sys
from rich.console import Console
from rich.text import Text
console = Console()
# Use emojis only if stdout supports unicode (not Windows cp1252)
try:
"\U0001f916".encode(sys.stdout.encoding or "utf-8")
server_icon, docs_icon = "\U0001f916", "\U0001f4da"
except (UnicodeEncodeError, LookupError):
server_icon, docs_icon = ">>", ">>"
art_lines = [
" _ _ _ ____ _ _ ____",
"| | | (_) _ \\ __ _| |_| |__ | _ \\ _____ __",
"| | | | | |_) / _` | __| '_ \\ | | | |/ _ \\ \\ / /",
"| |_| | | __/ (_| | |_| | | | | |_| | __/\\ V /",
" \\___/|_|_| \\__,_|\\__|_| |_| |____/ \\___| \\_/",
]
console.print()
for line in art_lines:
styled = Text(line)
styled.stylize("bold orange1")
console.print(styled)
console.print()
console.print(f" {server_icon} Server: [bold cyan]{base_url}[/bold cyan]")
console.print(
f" {docs_icon} Docs: [link=https://uipath.github.io/uipath-python/]"
"https://uipath.github.io/uipath-python/[/link]"
)
console.print()
console.print(
" [dim]This server is designed for development and testing.[/dim]"
)
console.print()
def _deferred_open_browser(self) -> None:
"""Open the browser after a short delay to let uvicorn bind."""
time.sleep(1.5)
webbrowser.open(f"http://{self.host}:{self.port}")