Files
sovereign-orchestrator/venv/lib/python3.12/site-packages/websockets/sync/utils.py
T
wmantly 70c71161f3 feat: initial orchestrator service with FastAPI backend and premium GUI
- FastAPI backend with full Proxmox VE API integration
- ISO builder using proxmox-auto-install-assistant
- Premium dark-mode SPA frontend with glassmorphism design
- VM lifecycle management (create, start, stop, destroy)
- Build pipeline tracking with real-time logs
- Deployment automation for custom auto-installer ISOs
- Production deployment script (setup.sh + systemd)
- Comprehensive README with API documentation
2026-06-21 22:57:32 -04:00

46 lines
1.1 KiB
Python

from __future__ import annotations
import time
__all__ = ["Deadline"]
class Deadline:
"""
Manage timeouts across multiple steps.
Args:
timeout: Time available in seconds or :obj:`None` if there is no limit.
"""
def __init__(self, timeout: float | None) -> None:
self.deadline: float | None
if timeout is None:
self.deadline = None
else:
self.deadline = time.monotonic() + timeout
def timeout(self, *, raise_if_elapsed: bool = True) -> float | None:
"""
Calculate a timeout from a deadline.
Args:
raise_if_elapsed: Whether to raise :exc:`TimeoutError`
if the deadline lapsed.
Raises:
TimeoutError: If the deadline lapsed.
Returns:
Time left in seconds or :obj:`None` if there is no limit.
"""
if self.deadline is None:
return None
timeout = self.deadline - time.monotonic()
if raise_if_elapsed and timeout <= 0:
raise TimeoutError("timed out")
return timeout