Coverage for scripts/_rhiza_yaml.py: 100%
197 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-31 06:18 +0000
1#!/usr/bin/env python3
2"""Minimal, dependency-free YAML reader/writer for rhiza template files.
4The bundled scripts (`status.py`, `validate.py`, `sync.py`, ...) are stdlib-only
5ports of the `rhiza` CLI's commands, so they can run inside this plugin without
6the CLI (or PyYAML) installed. They read `.rhiza/template.yml`,
7`.rhiza/template.lock`, and the upstream `.rhiza/template-bundles.yml`, and
8`sync.py` writes `.rhiza/template.lock`.
10`load_yaml` parses the subset of YAML those files use: nested mappings, block
11and inline (`[a, b]`) sequences, inline (`{source: x, dest: y}`) mappings, block
12scalars (`key: |`), quoted/bare scalars, and `#` comments. When PyYAML *is*
13importable we defer to it (same "stdlib works, third-party enhances" posture as
14the tomllib/tomli fallback), so hand-authored configs using constructs
15this parser doesn't cover still load correctly.
17`dump_yaml` emits the flat top-level scalar/sequence subset the lock file uses,
18matching PyYAML's `default_flow_style=False, sort_keys=False` layout (zero-indent
19list items, `[]` for empty lists, single-quoted values where a bare token would
20be re-read as a non-string) so a lock this module writes round-trips through all
21three readers (this parser, PyYAML, and the rhiza CLI).
23The built-in parser deliberately does NOT handle anchors, aliases, or multiple
24documents — none of which appear in rhiza template files.
25"""
27from __future__ import annotations
29import re
30from pathlib import Path
31from typing import Any
33try: # pragma: no cover - exercised only when PyYAML is installed
34 import yaml as _pyyaml
35except ModuleNotFoundError: # pragma: no cover
36 _pyyaml = None
38# YAML 1.1 timestamp shapes PyYAML resolves to datetime; we must quote these on
39# output (and never coerce them on input) to keep values like `synced_at` strings.
40_TIMESTAMP = re.compile(
41 r"^\d{4}-\d{1,2}-\d{1,2}([Tt ]\d{1,2}:\d{1,2}:\d{1,2}(\.\d+)?([Zz]|[+-]\d{1,2}(:\d{1,2})?)?)?$"
42)
43_BLOCK_SCALAR_INDICATORS = {"|", ">", "|-", ">-", "|+", ">+"}
46def load_yaml(path: Path) -> dict[str, Any]:
47 """Load a rhiza template/lock/bundles file into a plain dict.
49 Prefers PyYAML when available; otherwise falls back to the built-in
50 subset parser. A file whose top level is empty yields ``{}``. Raises
51 ``ValueError`` when the document's top level is not a mapping, mirroring
52 how the CLI treats a malformed config.
53 """
54 text = path.read_text(errors="ignore")
55 if _pyyaml is not None:
56 data = _pyyaml.safe_load(text)
57 if data is None:
58 return {}
59 if not isinstance(data, dict):
60 raise ValueError("top-level YAML is not a mapping")
61 return data
62 return _parse_subset(text)
65def dump_yaml(data: dict[str, Any], path: Path) -> None:
66 """Write *data* to *path* as YAML, matching PyYAML's block layout.
68 Only the flat subset the lock file uses is supported: top-level keys whose
69 values are scalars, ``None``, or lists of scalars. Nested mappings are not
70 emitted (the lock has none). The output re-reads identically via this
71 parser, PyYAML, and the rhiza CLI.
72 """
73 path.write_text(dumps_yaml(data))
76def dumps_yaml(data: dict[str, Any]) -> str:
77 """Serialise *data* to a YAML string (see :func:`dump_yaml`)."""
78 lines: list[str] = []
79 for key, value in data.items():
80 if isinstance(value, list):
81 if not value:
82 lines.append(f"{key}: []")
83 else:
84 lines.append(f"{key}:")
85 lines.extend(f"- {_emit_scalar(item)}" for item in value)
86 else:
87 lines.append(f"{key}: {_emit_scalar(value)}")
88 return "\n".join(lines) + "\n" if lines else ""
91def _emit_scalar(value: Any) -> str:
92 """Render a scalar for output, quoting when a bare token would misparse."""
93 if value is None:
94 return "null"
95 if isinstance(value, bool):
96 return "true" if value else "false"
97 if isinstance(value, int):
98 return str(value)
99 text = str(value)
100 if _needs_quote(text):
101 return "'" + text.replace("'", "''") + "'"
102 return text
105def _needs_quote(text: str) -> bool:
106 """Return True when *text* must be single-quoted to survive a round-trip."""
107 if text == "":
108 return True
109 if _scalar(text) != text:
110 # Would be re-read as bool/int/None/list/flow-map rather than a string.
111 return True
112 if _TIMESTAMP.match(text) or _is_float(text):
113 return True
114 if text[0] in "!&*?|>%@`\"'#[]{},":
115 return True
116 if text[0] in "-:" and (len(text) == 1 or text[1] == " "):
117 return True
118 return text != text.strip() or ": " in text or text.endswith(":") or "\n" in text
121def _is_float(text: str) -> bool:
122 """Return True when *text* parses as a float (and so needs quoting)."""
123 try:
124 float(text)
125 except ValueError:
126 return False
127 return True
130def _strip_comment(value: str) -> str:
131 """Drop a trailing ``# comment`` that sits outside any quotes."""
132 quote: str | None = None
133 for i, ch in enumerate(value):
134 if quote:
135 if ch == quote:
136 quote = None
137 elif ch in ("'", '"'):
138 quote = ch
139 elif ch == "#" and (i == 0 or value[i - 1] in " \t"):
140 return value[:i]
141 return value
144def _split_flow(inner: str) -> list[str]:
145 """Split the body of an inline ``[a, b, c]`` list on top-level commas."""
146 items: list[str] = []
147 buf = ""
148 quote: str | None = None
149 for ch in inner:
150 if quote:
151 buf += ch
152 if ch == quote:
153 quote = None
154 elif ch in ("'", '"'):
155 quote = ch
156 buf += ch
157 elif ch == ",":
158 items.append(buf)
159 buf = ""
160 else:
161 buf += ch
162 if buf.strip():
163 items.append(buf)
164 return items
167def _flow_map(inner: str) -> dict[str, Any]:
168 """Parse the body of an inline ``{a: x, b: y}`` mapping into a dict."""
169 result: dict[str, Any] = {}
170 for part in _split_flow(inner):
171 key, sep, rest = part.partition(":")
172 if sep:
173 result[key.strip()] = _scalar(rest.strip())
174 return result
177def _scalar(raw: str) -> Any:
178 """Coerce a scalar token to str/int/bool/None/list/dict, honouring quotes."""
179 s = raw.strip()
180 if not s:
181 return None
182 if (s[0] == '"' and s[-1] == '"') or (s[0] == "'" and s[-1] == "'"):
183 return s[1:-1]
184 if s.startswith("[") and s.endswith("]"):
185 body = s[1:-1].strip()
186 return [_scalar(x) for x in _split_flow(body)] if body else []
187 if s.startswith("{") and s.endswith("}"):
188 return _flow_map(s[1:-1].strip())
189 low = s.lower()
190 if low in ("null", "~"):
191 return None
192 if low in ("true", "false"):
193 return low == "true"
194 try:
195 return int(s)
196 except ValueError:
197 return s
200def _indent_of(line: str) -> int:
201 """Return the number of leading spaces on *line*."""
202 return len(line) - len(line.lstrip(" "))
205def _next_content(lines: list[str], i: int) -> int:
206 """Return the index of the next non-blank, non-comment line at or after *i*."""
207 while i < len(lines):
208 stripped = lines[i].strip()
209 if stripped and not stripped.startswith("#"):
210 return i
211 i += 1
212 return len(lines)
215def _parse_subset(text: str) -> dict[str, Any]:
216 """Parse the nested scalar/list/mapping YAML subset rhiza files use."""
217 lines = text.splitlines()
218 value, _ = _parse_map(lines, _next_content(lines, 0), 0)
219 return value
222def _parse_map(lines: list[str], i: int, indent: int) -> tuple[dict[str, Any], int]:
223 """Parse a block mapping whose keys sit at *indent*, returning it and the next index."""
224 data: dict[str, Any] = {}
225 while True:
226 i = _next_content(lines, i)
227 if i >= len(lines) or _indent_of(lines[i]) < indent:
228 break
229 stripped = lines[i].strip()
230 if stripped.startswith("- ") or stripped == "-":
231 break # a sequence at this level is not part of a mapping
232 if ":" not in stripped:
233 i += 1 # tolerate a stray non-mapping line, as the CLI reader does
234 continue
235 key, _, rest = stripped.partition(":")
236 key = key.strip()
237 rest = _strip_comment(rest).strip()
238 i += 1
239 if rest in _BLOCK_SCALAR_INDICATORS:
240 data[key], i = _parse_block_scalar(lines, i, indent)
241 elif rest == "":
242 data[key], i = _parse_child(lines, i, indent)
243 else:
244 data[key] = _scalar(rest)
245 return data, i
248def _parse_child(lines: list[str], i: int, parent_indent: int) -> tuple[Any, int]:
249 """Parse the value introduced by a bare ``key:`` line, or ``None`` when absent."""
250 j = _next_content(lines, i)
251 if j >= len(lines):
252 return None, i
253 child_indent = _indent_of(lines[j])
254 child = lines[j].strip()
255 is_seq = child.startswith("- ") or child == "-"
256 # Block sequences may sit at the parent's indent (zero-indent style); block
257 # mappings must be strictly deeper.
258 if is_seq and child_indent >= parent_indent:
259 return _parse_seq(lines, j, child_indent)
260 if not is_seq and child_indent > parent_indent:
261 return _parse_map(lines, j, child_indent)
262 return None, i
265def _parse_seq(lines: list[str], i: int, indent: int) -> tuple[list[Any], int]:
266 """Parse a block sequence whose ``- `` items sit at *indent*."""
267 items: list[Any] = []
268 while True:
269 i = _next_content(lines, i)
270 if i >= len(lines) or _indent_of(lines[i]) < indent:
271 break
272 stripped = lines[i].strip()
273 if not (stripped.startswith("- ") or stripped == "-"):
274 break
275 item = "" if stripped == "-" else stripped[2:]
276 item = _strip_comment(item).strip()
277 if item and item[0] not in "[{'\"" and re.match(r"[^:\s]+:(\s|$)", item):
278 # A block mapping under this item: reparse from the "- " column.
279 lines[i] = lines[i].replace("- ", " ", 1)
280 value, i = _parse_map(lines, i, _indent_of(lines[i]))
281 items.append(value)
282 else:
283 items.append(_scalar(item))
284 i += 1
285 return items, i
288def _parse_block_scalar(lines: list[str], i: int, parent_indent: int) -> tuple[str, int]:
289 """Consume the indented body of a ``key: |`` block scalar into a string."""
290 body: list[str] = []
291 while i < len(lines):
292 line = lines[i]
293 if line.strip() and _indent_of(line) <= parent_indent:
294 break
295 body.append(line.strip())
296 i += 1
297 return "\n".join(body).strip(), i
300def as_list(value: Any) -> list[str]:
301 """Normalise a scalar/None/list config field into a list of strings."""
302 if value is None:
303 return []
304 if isinstance(value, list):
305 return [str(x) for x in value]
306 if isinstance(value, str):
307 parts = value.split("\\n") if "\\n" in value and "\n" not in value else value.split("\n")
308 return [p.strip() for p in parts if p.strip()]
309 return [str(value)]