mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: add docstrings to all API endpoints and daemon handlers

Browse Source 
Add comprehensive docstrings to firewall, DHCP, proxy, wireguard, certs,
and logs API endpoints. Document parameters, return values, and error cases
for the documentation system.
This commit is contained in:
Mike Teehan
2026-05-30 16:15:45 +00:00
parent bd98830638
commit 2f215793e9
17 changed files with 1550 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/state.py
+129 -7
View File
@@ -40,23 +40,53 @@ class State:
Each subsystem's value is a dict collected from the corresponding
``collect_*`` function. A value of ``None`` means the subsystem has
not been populated yet or the last collection failed.
Attributes:
SUBSYSTEMS: Ordered list of subsystem names.
_data: Dict mapping subsystem names to their state data.
"""
SUBSYSTEMS: ClassVar[list[str]] = ["firewall", "dnsmasq", "nginx", "acme", "wireguard"]
SUBSYSTEMS: ClassVar[list[str]] = [
"firewall",
"dnsmasq",
"nginx",
"acme",
"wireguard",
]
def __init__(self) -> None:
"""Initialize the state store with empty subsystem slots."""
self._data: dict[str, dict[str, Any] | None] = {
name: None for name in self.SUBSYSTEMS
}
def get(self, subsystem: str) -> dict[str, Any] | None:
"""Get state data for *subsystem*.
Args:
subsystem: Subsystem name.
Returns:
State dict, or ``None`` if not populated.
"""
return self._data.get(subsystem)
def set(self, subsystem: str, data: dict[str, Any] | None) -> None:
"""Set state data for *subsystem*.
Args:
subsystem: Subsystem name.
data: State data, or ``None`` to clear.
"""
self._data[subsystem] = data
def populate(self, subsystems: list[str] | None = None) -> None:
"""Collect state for *subsystems* (all if None)."""
"""Collect state for *subsystems* (all if ``None``).
Args:
subsystems: List of subsystem names to collect. Collects all
subsystems when ``None``.
"""
targets = subsystems or self.SUBSYSTEMS
for name in targets:
collector = _COLLECTORS.get(name)
@@ -73,6 +103,11 @@ class State:
self._data[name] = None
def is_populated(self) -> bool:
"""Check whether all subsystem states have been populated.
Returns:
``True`` if every subsystem has non-``None`` state data.
"""
return all(v is not None for v in self._data.values())
@@ -88,6 +123,15 @@ _COLLECTORS: dict[str, Any] = {}
def register_collector(subsystem: str, fn: Any) -> Any:
"""Register *fn* as the state collector for *subsystem*.
Args:
subsystem: Subsystem name to register for.
fn: Collector function to register.
Returns:
The *fn* function (for decorator usage).
"""
_COLLECTORS[subsystem] = fn
return fn
@@ -98,10 +142,19 @@ def register_collector(subsystem: str, fn: Any) -> Any:
def _now_iso() -> str:
"""Return the current UTC time as an ISO 8601 string."""
return datetime.now(UTC).isoformat()
def _fp_to_str(fp: dict[str, Any]) -> str:
"""Convert a port-forward dict to a compact string representation.
Args:
fp: Port-forward entry containing port and proto keys.
Returns:
Comma-separated string of key=value pairs (e.g. ``port=443,proto=tcp``).
"""
parts = [f"port={fp['port']}", f"proto={fp['proto']}"]
if "toaddr" in fp:
parts.append(f"toaddr={fp['toaddr']}")
@@ -111,7 +164,12 @@ def _fp_to_str(fp: dict[str, Any]) -> str:
def _collect_firewall() -> dict[str, Any]:
"""Return the complete current state of firewalld."""
"""Return the complete current state of firewalld.
Returns:
Dict containing firewall zones, interfaces, rules, config, and
pending changes.
"""
zone_names = run(["firewall-cmd", "--get-zones"], sudo=True).split()
active_raw = run(["firewall-cmd", "--get-active-zones"], sudo=True)
active = _parse_active_zones(active_raw)
@@ -221,7 +279,11 @@ register_collector("firewall", _collect_firewall)
def _collect_dnsmasq() -> dict[str, Any]:
"""Collect dnsmasq status, config, and leases."""
"""Collect dnsmasq status, config, and leases.
Returns:
Dict containing config, service status, leases, and timestamp.
"""
CONFIG_DIR = PROJECT_DIR / "config" / "dnsmasq"
CONFIG_PATH = CONFIG_DIR / "config.json"
DNSMASQ_CONF = "/etc/dnsmasq.d/vacuum-wall.conf"
@@ -311,7 +373,11 @@ register_collector("dnsmasq", _collect_dnsmasq)
def _collect_nginx() -> dict[str, Any]:
"""Collect nginx config and domains list."""
"""Collect nginx config and domains list.
Returns:
Dict containing config, domains, and timestamp.
"""
CONFIG_DIR = PROJECT_DIR / "config" / "nginx"
CONFIG_FILE = CONFIG_DIR / "config.json"
SITES_DIR = PROJECT_DIR / "data" / "nginx" / "sites-enabled"
@@ -381,6 +447,14 @@ register_collector("nginx", _collect_nginx)
def _find_acme() -> str:
"""Locate the ``acme.sh`` binary on the filesystem.
Returns:
Absolute path to the ``acme.sh`` executable.
Raises:
FileNotFoundError: If acme.sh cannot be found.
"""
acme_home = PROJECT_DIR / "data" / "acme"
candidates = [acme_home / "acme.sh", Path("/usr/local/bin/acme.sh")]
for path in candidates:
@@ -393,6 +467,17 @@ def _find_acme() -> str:
def _run_acme(args: list[str]) -> str:
"""Run ``acme.sh`` with *args* and return combined output.
Args:
args: Command-line arguments to pass after the home/config flags.
Returns:
Combined stdout/stderr output.
Raises:
RuntimeError: If acme.sh exits non-zero or times out.
"""
acme_bin = _find_acme()
acme_home_env = os.environ.get("ACME_HOME", str(PROJECT_DIR / "data" / "acme"))
cmd = [acme_bin, "--home", acme_home_env, "--config-home", acme_home_env, *args]
@@ -421,6 +506,14 @@ def _run_acme(args: list[str]) -> str:
def _days_until(date_str: str) -> int | None:
"""Parse a date string and return days until *date_str* from now.
Args:
date_str: Date string in common ACME formats.
Returns:
Number of days remaining, or ``None`` if empty or unparseable.
"""
if not date_str:
return None
for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"):
@@ -438,6 +531,14 @@ def _days_until(date_str: str) -> int | None:
def _parse_acme_list_output(raw: str) -> list[dict]:
"""Parse ``acme.sh --list`` output into a list of certificate dicts.
Args:
raw: Raw output string from ``acme.sh --list``.
Returns:
List of dicts with certificate entry fields.
"""
entries: list[dict] = []
for line in raw.strip().splitlines():
line = line.strip()
@@ -455,11 +556,24 @@ def _parse_acme_list_output(raw: str) -> list[dict]:
def _has_auto_renew(domain: str) -> bool:
"""Check whether *domain* has an auto-renew configuration file.
Args:
domain: Domain name to check.
Returns:
``True`` if a corresponding ``acme.sh`` config file exists.
"""
acme_home_env = os.environ.get("ACME_HOME", str(PROJECT_DIR / "data" / "acme"))
return bool(Path(acme_home_env) / f"{domain}.conf")
def _get_acme_email() -> str:
"""Read the ACME ``acme.sh`` email from the account config file.
Returns:
Email string, or empty string if not found.
"""
acme_home_default = str(PROJECT_DIR / "data" / "acme")
try:
acme_home = Path(os.environ.get("ACME_HOME", acme_home_default))
@@ -475,7 +589,11 @@ def _get_acme_email() -> str:
def _collect_acme() -> dict[str, Any]:
"""Collect ACME certificate list and email."""
"""Collect ACME certificate list and email.
Returns:
Dict containing certificate details and registered email.
"""
email = _get_acme_email()
certs: list[dict[str, Any]] = []
@@ -529,7 +647,11 @@ register_collector("acme", _collect_acme)
def _collect_wireguard() -> dict[str, Any]:
"""Collect WireGuard config, status, and peers."""
"""Collect WireGuard config, status, and peers.
Returns:
Dict containing interface config, runtime status, and peers.
"""
CONFIG_PATH = PROJECT_DIR / "config" / "wireguard" / "config.json"
DEFAULT_CONFIG: dict[str, Any] = {