# YAML-access python import yaml from pathlib import Path from typing import Any, Union PROJECT_PATH = Path(__file__).resolve().parent PathLike = Union[str, Path, tuple[str, ...]] # ============================================================ # 1. Dot-access wrapper # ============================================================ class YAMLNode: def __init__(self, data: Any): object.__setattr__(self, "_data", data) # ============================================================ # Core access # ============================================================ def __getitem__(self, key): return self._wrap(self._data[key]) def __getattr__(self, key): if isinstance(self._data, dict) and key in self._data: return self._wrap(self._data[key]) raise AttributeError(f"Key not found: {key}") def __setitem__(self, key, value): if isinstance(self._data, dict): self._data[key] = self._unwrap(value) return if isinstance(self._data, list): self._data[key] = self._unwrap(value) return raise TypeError(f"Cannot set item on {type(self._data).__name__}") def __setattr__(self, key, value): if key == "_data": object.__setattr__(self, key, value) return if isinstance(self._data, dict): self._data[key] = self._unwrap(value) return raise AttributeError("Cannot set attribute on non-dict YAML node") # ============================================================ # Dict-like methods # ============================================================ def get(self, key, default=None): if isinstance(self._data, dict): return self._wrap(self._data.get(key, default)) return default def items(self): if not isinstance(self._data, dict): raise TypeError(f"items() is only available on dict nodes, got {type(self._data).__name__}") for key, value in self._data.items(): yield key, self._wrap(value) def keys(self): if not isinstance(self._data, dict): raise TypeError(f"keys() is only available on dict nodes, got {type(self._data).__name__}") return self._data.keys() def values(self): if not isinstance(self._data, dict): raise TypeError(f"values() is only available on dict nodes, got {type(self._data).__name__}") for value in self._data.values(): yield self._wrap(value) def update(self, other=None, **kwargs): if not isinstance(self._data, dict): raise TypeError(f"update() is only available on dict nodes, got {type(self._data).__name__}") if other: if isinstance(other, YAMLNode): other = other.raw() for key, value in dict(other).items(): self._data[key] = self._unwrap(value) for key, value in kwargs.items(): self._data[key] = self._unwrap(value) return self # ============================================================ # List-like helpers # ============================================================ def append(self, value): if not isinstance(self._data, list): raise TypeError(f"append() is only available on list nodes, got {type(self._data).__name__}") self._data.append(self._unwrap(value)) return self def extend(self, values): if not isinstance(self._data, list): raise TypeError(f"extend() is only available on list nodes, got {type(self._data).__name__}") if isinstance(values, YAMLNode): values = values.raw() self._data.extend([self._unwrap(v) for v in values]) return self # ============================================================ # Python protocol support # ============================================================ def __iter__(self): if isinstance(self._data, dict): return iter(self._data) if isinstance(self._data, list): return (self._wrap(value) for value in self._data) raise TypeError(f"{type(self._data).__name__} node is not iterable") def __len__(self): if isinstance(self._data, (dict, list, tuple, str)): return len(self._data) raise TypeError(f"{type(self._data).__name__} node has no len()") def __bool__(self): return bool(self._data) def __contains__(self, key): if isinstance(self._data, dict): return key in self._data if isinstance(self._data, list): return key in self._data return False # ============================================================ # Conversion # ============================================================ def raw(self): return self._data def to_dict(self): return self._deep_unwrap(self._data) @staticmethod def _wrap(value): if isinstance(value, (dict, list)): return YAMLNode(value) return value @staticmethod def _unwrap(value): if isinstance(value, YAMLNode): return value.raw() return value @classmethod def _deep_unwrap(cls, value): if isinstance(value, YAMLNode): return cls._deep_unwrap(value.raw()) if isinstance(value, dict): return {key: cls._deep_unwrap(val) for key, val in value.items()} if isinstance(value, list): return [cls._deep_unwrap(item) for item in value] return value def __repr__(self): return repr(self._data) # ============================================================ # 2. Path resolving # ============================================================ def _project_path(path: PathLike = PROJECT_PATH) -> Path: if isinstance(path, tuple): return PROJECT_PATH.joinpath(*path) path = Path(path) # "/seeds/planner" should mean relative to PROJECT_PATH, # not system root. raw = str(path) if raw.startswith("/"): raw = raw.lstrip("/") return PROJECT_PATH / raw if path.is_absolute(): return path return PROJECT_PATH / path def _possible_yaml_paths(base: Path): if base.suffix in [".yaml", ".yml"]: yield base else: yield base.with_suffix(".yaml") yield base.with_suffix(".yml") yield base def _find_yaml_file_and_inner(path: PathLike): """ Examples: /seeds/planner -> .project/seeds/planner.yaml, inner=None /seeds/planner/final_export.io.dasd -> .project/seeds/planner.yaml, inner="final_export.io.dasd" /seeds/planner/final_export/io/dasd -> .project/seeds/planner.yaml, inner="final_export/io/dasd" """ full = _project_path(path) # Direct file check. for candidate in _possible_yaml_paths(full): if candidate.exists() and candidate.is_file(): return candidate, None parts = full.parts project_parts = PROJECT_PATH.parts # Remove project prefix from parts. if parts[:len(project_parts)] == project_parts: rel_parts = parts[len(project_parts):] else: rel_parts = parts # Try longest possible YAML file path. # /seeds/planner/final_export/io # tries: # seeds/planner/final_export/io.yaml # seeds/planner/final_export.yaml # seeds/planner.yaml # seeds.yaml for i in range(len(rel_parts), 0, -1): file_candidate = PROJECT_PATH.joinpath(*rel_parts[:i]) for yaml_candidate in _possible_yaml_paths(file_candidate): if yaml_candidate.exists() and yaml_candidate.is_file(): inner_parts = rel_parts[i:] inner = "/".join(inner_parts) if inner_parts else None return yaml_candidate, inner raise FileNotFoundError(f"No YAML file found for path: {path}") # ============================================================ # 3. Inner key parser # ============================================================ def _parse_inner_path(inner: str | None): """ Supports: final_export.io.dasd final_export['io']['dasd'] final_export["io"].dasd final_export/io/dasd items[0].name """ if not inner: return [] tokens = [] i = 0 n = len(inner) while i < n: ch = inner[i] if ch in ". /": i += 1 continue if ch == "[": i += 1 # Quoted key: ['abc'] or ["abc"] if i < n and inner[i] in ["'", '"']: quote = inner[i] i += 1 start = i while i < n and inner[i] != quote: i += 1 if i >= n: raise ValueError(f"Unclosed quoted bracket in path: {inner}") tokens.append(inner[start:i]) i += 1 if i >= n or inner[i] != "]": raise ValueError(f"Missing closing ] in path: {inner}") i += 1 continue # Numeric index or bare key: [0] or [name] start = i while i < n and inner[i] != "]": i += 1 if i >= n: raise ValueError(f"Unclosed bracket in path: {inner}") raw = inner[start:i].strip() if raw.isdigit(): tokens.append(int(raw)) else: tokens.append(raw) i += 1 continue # Bare token: abc until ., /, or [ start = i while i < n and inner[i] not in "./[": i += 1 raw = inner[start:i].strip() if raw: tokens.append(raw) return tokens # ============================================================ # 4. Walking YAML data # ============================================================ def _get_from_tokens(data: Any, tokens: list): cur = data for token in tokens: if isinstance(cur, YAMLNode): cur = cur.raw() if isinstance(token, int): if not isinstance(cur, list): raise TypeError(f"Expected list for index [{token}], got {type(cur).__name__}") cur = cur[token] else: if not isinstance(cur, dict): raise TypeError(f"Expected dict for key '{token}', got {type(cur).__name__}") cur = cur[token] return cur def _set_from_tokens(data: Any, tokens: list, value: Any): if not tokens: return value cur = data for idx, token in enumerate(tokens[:-1]): next_token = tokens[idx + 1] if isinstance(token, int): if not isinstance(cur, list): raise TypeError(f"Expected list for index [{token}], got {type(cur).__name__}") while len(cur) <= token: cur.append({} if not isinstance(next_token, int) else []) if cur[token] is None: cur[token] = {} if not isinstance(next_token, int) else [] cur = cur[token] else: if not isinstance(cur, dict): raise TypeError(f"Expected dict for key '{token}', got {type(cur).__name__}") if token not in cur or cur[token] is None: cur[token] = {} if not isinstance(next_token, int) else [] cur = cur[token] last = tokens[-1] if isinstance(last, int): if not isinstance(cur, list): raise TypeError(f"Expected list for final index [{last}], got {type(cur).__name__}") while len(cur) <= last: cur.append(None) cur[last] = YAMLNode._unwrap(value) else: if not isinstance(cur, dict): raise TypeError(f"Expected dict for final key '{last}', got {type(cur).__name__}") cur[last] = YAMLNode._unwrap(value) return data # ============================================================ # 5. Public API # ============================================================ def ExploreFromPath(path: PathLike = PROJECT_PATH): yaml_file, inner = _find_yaml_file_and_inner(path) tokens = _parse_inner_path(inner) return { "yaml_file": yaml_file, "inner": inner, "tokens": tokens, } def Get(path: PathLike = PROJECT_PATH, *, raw: bool = False): yaml_file, inner = _find_yaml_file_and_inner(path) tokens = _parse_inner_path(inner) with open(yaml_file, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} value = _get_from_tokens(data, tokens) if raw: return value return YAMLNode._wrap(value) def Set(path: PathLike, value: Any): yaml_file, inner = _find_yaml_file_and_inner(path) tokens = _parse_inner_path(inner) with open(yaml_file, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} data = _set_from_tokens(data, tokens, value) with open(yaml_file, "w", encoding="utf-8") as f: yaml.safe_dump(data, f, sort_keys=False, allow_unicode=True) return YAMLNode._wrap(data)