74 lines
2 KiB
Python
74 lines
2 KiB
Python
|
"""State file management for OpsBridge."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
from bridge.models import BridgeState
|
||
|
|
|
||
|
|
|
||
|
|
def _default_state_dir() -> Path:
|
||
|
|
return Path.home() / ".local" / "state" / "bridge"
|
||
|
|
|
||
|
|
|
||
|
|
class StateManager:
|
||
|
|
def __init__(self, state_dir: Optional[Path] = None):
|
||
|
|
self._dir = Path(state_dir) if state_dir else _default_state_dir()
|
||
|
|
|
||
|
|
def _ensure_dir(self) -> None:
|
||
|
|
self._dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
def _state_path(self, name: str) -> Path:
|
||
|
|
return self._dir / f"{name}.state"
|
||
|
|
|
||
|
|
def _pid_path(self, name: str) -> Path:
|
||
|
|
return self._dir / f"{name}.pid"
|
||
|
|
|
||
|
|
def read_state(self, name: str) -> BridgeState:
|
||
|
|
path = self._state_path(name)
|
||
|
|
if not path.exists():
|
||
|
|
return BridgeState.STOPPED
|
||
|
|
text = path.read_text().strip()
|
||
|
|
try:
|
||
|
|
return BridgeState(text)
|
||
|
|
except ValueError:
|
||
|
|
return BridgeState.STOPPED
|
||
|
|
|
||
|
|
def write_state(self, name: str, state: BridgeState) -> None:
|
||
|
|
self._ensure_dir()
|
||
|
|
self._state_path(name).write_text(state.value)
|
||
|
|
|
||
|
|
def read_pid(self, name: str) -> Optional[int]:
|
||
|
|
path = self._pid_path(name)
|
||
|
|
if not path.exists():
|
||
|
|
return None
|
||
|
|
try:
|
||
|
|
pid = int(path.read_text().strip())
|
||
|
|
except (ValueError, OSError):
|
||
|
|
return None
|
||
|
|
if _pid_alive(pid):
|
||
|
|
return pid
|
||
|
|
return None
|
||
|
|
|
||
|
|
def write_pid(self, name: str, pid: int) -> None:
|
||
|
|
self._ensure_dir()
|
||
|
|
self._pid_path(name).write_text(str(pid))
|
||
|
|
|
||
|
|
def clear_pid(self, name: str) -> None:
|
||
|
|
path = self._pid_path(name)
|
||
|
|
if path.exists():
|
||
|
|
path.unlink()
|
||
|
|
|
||
|
|
def is_running(self, name: str) -> bool:
|
||
|
|
return self.read_pid(name) is not None
|
||
|
|
|
||
|
|
|
||
|
|
def _pid_alive(pid: int) -> bool:
|
||
|
|
"""Return True if the process with given PID exists."""
|
||
|
|
try:
|
||
|
|
os.kill(pid, 0)
|
||
|
|
return True
|
||
|
|
except (ProcessLookupError, PermissionError):
|
||
|
|
return False
|