Spaces:
Build error
Build error
| """Providers for launching ASGI applications via ``uv run``.""" | |
| from __future__ import annotations | |
| import os | |
| import socket | |
| import subprocess | |
| import time | |
| from typing import Dict, Optional | |
| import requests | |
| from .providers import RuntimeProvider | |
| def _check_uv_installed() -> None: | |
| try: | |
| subprocess.check_output(["uv", "--version"]) | |
| except FileNotFoundError as exc: | |
| raise RuntimeError( | |
| "`uv` executable not found. Install uv from https://docs.astral.sh and ensure it is on PATH." | |
| ) from exc | |
| def _find_free_port() -> int: | |
| with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: | |
| sock.bind(("", 0)) | |
| sock.listen(1) | |
| return sock.getsockname()[1] | |
| def _create_uv_command( | |
| *, | |
| host: str, | |
| port: int, | |
| reload: bool, | |
| workers: int, | |
| app: str, | |
| project_path: str, | |
| ) -> list[str]: | |
| command: list[str] = ["uv", "run", "--isolated", "--project", project_path] | |
| command.append("--") | |
| command.extend( | |
| [ | |
| "uvicorn", | |
| app, | |
| "--host", | |
| host, | |
| "--port", | |
| str(port), | |
| "--workers", | |
| str(workers), | |
| ] | |
| ) | |
| if reload: | |
| command.append("--reload") | |
| return command | |
| def _poll_health(health_url: str, timeout_s: float) -> None: | |
| """Poll a health endpoint until it returns HTTP 200 or times out.""" | |
| deadline = time.time() + timeout_s | |
| while time.time() < deadline: | |
| try: | |
| timeout = max(0.0001, min(deadline - time.time(), 2.0)) | |
| response = requests.get(health_url, timeout=timeout) | |
| if response.status_code == 200: | |
| return | |
| except requests.RequestException: | |
| continue | |
| time.sleep(0.5) | |
| raise TimeoutError(f"Server did not become ready within {timeout_s:.1f} seconds") | |
| class UVProvider(RuntimeProvider): | |
| """ | |
| RuntimeProvider implementation backed by ``uv run``. | |
| Args: | |
| project_path: Local path to a uv project (passed to ``uv run --project``) | |
| app: ASGI application path for uvicorn (defaults to ``server.app:app``) | |
| host: Host interface to bind to (defaults to ``0.0.0.0``) | |
| reload: Whether to enable uvicorn's reload mode | |
| env_vars: Environment variables to pass through to the spawned process | |
| context_timeout_s: How long to wait for the environment to become ready | |
| Example: | |
| >>> provider = UVProvider(project_path="/path/to/env") | |
| >>> base_url = provider.start() | |
| >>> print(base_url) # http://localhost:8000 | |
| >>> # Use the environment via base_url | |
| >>> provider.stop() | |
| """ | |
| def __init__( | |
| self, | |
| *, | |
| project_path: str, | |
| app: str = "server.app:app", | |
| host: str = "0.0.0.0", | |
| reload: bool = False, | |
| env_vars: Optional[Dict[str, str]] = None, | |
| context_timeout_s: float = 60.0, | |
| ): | |
| """Initialize the UVProvider.""" | |
| self.project_path = os.path.abspath(project_path) | |
| self.app = app | |
| self.host = host | |
| self.reload = reload | |
| self.env_vars = env_vars | |
| self.context_timeout_s = context_timeout_s | |
| _check_uv_installed() | |
| self._process = None | |
| self._base_url = None | |
| def start( | |
| self, | |
| port: Optional[int] = None, | |
| env_vars: Optional[Dict[str, str]] = None, | |
| workers: int = 1, | |
| **_: Dict[str, str], | |
| ) -> str: | |
| """ | |
| Start the environment via `uv run`. | |
| Args: | |
| port: The port to bind the environment to | |
| env_vars: Environment variables to pass to the environment | |
| workers: The number of workers to use | |
| Returns: | |
| The base URL of the environment | |
| Raises: | |
| RuntimeError: If the environment is already running | |
| """ | |
| if self._process is not None and self._process.poll() is None: | |
| raise RuntimeError("UVProvider is already running") | |
| bind_port = port or _find_free_port() | |
| command = _create_uv_command( | |
| host=self.host, | |
| port=bind_port, | |
| reload=self.reload, | |
| workers=workers, | |
| app=self.app, | |
| project_path=self.project_path, | |
| ) | |
| env = os.environ.copy() | |
| if self.env_vars: | |
| env.update(self.env_vars) | |
| if env_vars: | |
| env.update(env_vars) | |
| try: | |
| self._process = subprocess.Popen(command, env=env) | |
| except OSError as exc: | |
| raise RuntimeError(f"Failed to launch `uv run`: {exc}") from exc | |
| client_host = "127.0.0.1" if self.host in {"0.0.0.0", "::"} else self.host | |
| self._base_url = f"http://{client_host}:{bind_port}" | |
| return self._base_url | |
| def wait_for_ready(self, timeout_s: float = 60.0) -> None: | |
| """ | |
| Wait for the environment to become ready. | |
| Args: | |
| timeout_s: The timeout to wait for the environment to become ready | |
| Raises: | |
| RuntimeError: If the environment is not running | |
| TimeoutError: If the environment does not become ready within the timeout | |
| """ | |
| if self._process and self._process.poll() is not None: | |
| code = self._process.returncode | |
| raise RuntimeError(f"uv process exited prematurely with code {code}") | |
| _poll_health(f"{self._base_url}/health", timeout_s=timeout_s) | |
| def stop(self) -> None: | |
| """ | |
| Stop the environment. | |
| Raises: | |
| RuntimeError: If the environment is not running | |
| """ | |
| if self._process is None: | |
| return | |
| if self._process.poll() is None: | |
| self._process.terminate() | |
| try: | |
| self._process.wait(timeout=10.0) | |
| except subprocess.TimeoutExpired: | |
| self._process.kill() | |
| self._process.wait(timeout=5.0) | |
| self._process = None | |
| self._base_url = None | |
| def base_url(self) -> str: | |
| """ | |
| The base URL of the environment. | |
| Returns: | |
| The base URL of the environment | |
| Raises: | |
| RuntimeError: If the environment is not running | |
| """ | |
| if self._base_url is None: | |
| raise RuntimeError("UVProvider has not been started") | |
| return self._base_url | |