Coverage for sygaldry / codegen.py: 89%
304 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-01 00:13 -0400
1from __future__ import annotations
3__author__ = "Rohan B. Dalton"
5import re
6from dataclasses import dataclass, field
7from typing import Any, Optional
9from .types import RESERVED_KEYS
11_NON_IDENTIFIER_RE = re.compile(r"[^a-zA-Z0-9_]")
14@dataclass(frozen=True)
15class SourceMapping:
16 """
17 Map a generated line number to a config path.
18 """
20 line: int
21 config_path: str
24@dataclass
25class ComponentEntry:
26 """
27 A typed component extracted from the config.
28 """
30 config_path: str
31 var_name: str
32 import_path: str
33 args: list[Any] = field(default_factory=list)
34 kwargs: dict[str, Any] = field(default_factory=dict)
35 is_func: bool = False
38@dataclass
39class PlainEntry:
40 """
41 A plain (non-typed) config value.
42 """
44 config_path: str
45 var_name: str
46 value: Any
49@dataclass
50class RefExpression:
51 """
52 A reference expression to be emitted as a variable or attribute access.
53 """
55 target: str
58@dataclass
59class EntriesExpression:
60 """
61 A dict built from _entries.
62 """
64 items: list[tuple[Any, Any]]
67@dataclass
68class AnalysisResult:
69 """
70 Result of analyzing a config dict.
71 """
73 components: list[ComponentEntry] = field(default_factory=list)
74 plains: list[PlainEntry] = field(default_factory=list)
75 topological_order: list[str] = field(default_factory=list)
76 imports: dict[str, tuple[str, str]] = field(default_factory=dict)
79def _to_identifier(key: str) -> str:
80 """
81 Convert a config key to a valid Python identifier.
82 """
83 if result := _NON_IDENTIFIER_RE.sub("_", key):
84 if result[0].isdigit():
85 result = f"_{result}"
86 return result
87 else:
88 return "_unnamed"
91def _split_import_path(dotted: str) -> tuple[str, str]:
92 """
93 Split 'a.b.ClassName' into ('a.b', 'ClassName').
94 """
95 parts = dotted.rsplit(".", 1)
96 if len(parts) == 1:
97 return parts[0], parts[0]
98 else:
99 return parts[0], parts[1]
102class ConfigAnalyzer:
103 """
104 Walk a loaded config dict and extract typed component specs.
105 """
107 def __init__(self, config: dict[str, Any]) -> None:
108 self._config = config
109 self._components: list[ComponentEntry] = list()
110 self._plains: list[PlainEntry] = list()
111 self._imports: dict[str, tuple[str, str]] = dict()
112 self._deps: dict[str, list[str]] = dict()
113 self._variable_names: dict[str, str] = dict()
114 self._anon_counter = 0
115 self._used_names: set[str] = set()
117 def analyze(self) -> AnalysisResult:
118 """
119 Analyze the config and return structured results.
120 """
121 for key, value in self._config.items():
122 self._analyze_top_level(key, value)
123 order = self._topological_sort()
124 result = AnalysisResult(
125 components=self._components,
126 plains=self._plains,
127 topological_order=order,
128 imports=self._imports,
129 )
130 return result
132 def _alloc_name(self, preferred: str) -> str:
133 """
134 Allocate a unique variable name.
135 """
136 name = _to_identifier(preferred)
137 if name not in self._used_names: 137 ↛ 141line 137 didn't jump to line 141 because the condition on line 137 was always true
138 self._used_names.add(name)
139 return name
140 else:
141 counter = 0
142 while f"{name}_{counter}" in self._used_names:
143 counter += 1
144 unique = f"{name}_{counter}"
145 self._used_names.add(unique)
146 return unique
148 def _alloc_anon(self, hint: str) -> str:
149 """
150 Allocate an anonymous variable name.
151 """
152 name = f"__sygaldry_{_to_identifier(hint)}_{self._anon_counter}"
153 self._anon_counter += 1
154 self._used_names.add(name)
155 return name
157 def _analyze_top_level(self, key: str, value: Any) -> None:
158 """
159 Analyze a single top-level config entry.
160 """
161 variable_name = self._alloc_name(key)
162 self._variable_names[key] = variable_name
163 self._deps[key] = list()
165 if isinstance(value, dict):
166 if "_ref" in value:
167 ref_target = value["_ref"]
168 top = ref_target.split(".")[0]
169 self._deps[key].append(top)
170 self._plains.append(
171 PlainEntry(
172 config_path=key,
173 var_name=variable_name,
174 value=RefExpression(target=ref_target),
175 )
176 )
177 elif "_type" in value:
178 self._analyze_type_entry(key, variable_name, value)
179 elif "_func" in value:
180 self._analyze_func_entry(key, variable_name, value)
181 elif value.get("_deep") is False:
182 self._plains.append(PlainEntry(config_path=key, var_name=variable_name, value=...))
183 elif "_entries" in value:
184 entries_expr = self._analyze_entries(value["_entries"], key, self._deps[key])
185 self._plains.append(
186 PlainEntry(config_path=key, var_name=variable_name, value=entries_expr)
187 )
188 else:
189 self._plains.append(
190 PlainEntry(config_path=key, var_name=variable_name, value=value)
191 )
192 else:
193 self._plains.append(PlainEntry(config_path=key, var_name=variable_name, value=value))
195 def _analyze_type_entry(
196 self, config_path: str, var_name: str, entry: dict[str, Any]
197 ) -> None:
198 """
199 Analyze a _type entry.
200 """
201 import_path = entry["_type"]
202 mod, name = _split_import_path(import_path)
203 self._imports[import_path] = (mod, name)
205 raw_args = entry.get("_args", [])
206 args = list()
207 for idx, arg in enumerate(raw_args):
208 processed = self._process_arg_value(
209 arg, f"{config_path}._args.{idx}", self._deps.setdefault(config_path, [])
210 )
211 args.append(processed)
213 kwargs: dict[str, Any] = dict()
214 for key, value in entry.items():
215 if key in RESERVED_KEYS:
216 continue
217 processed = self._process_arg_value(
218 value, f"{config_path}.{key}", self._deps.setdefault(config_path, [])
219 )
220 kwargs[key] = processed
222 raw_kwargs = entry.get("_kwargs", {})
223 for key, value in raw_kwargs.items():
224 processed = self._process_arg_value(
225 value, f"{config_path}._kwargs.{key}", self._deps.setdefault(config_path, [])
226 )
227 kwargs[key] = processed
229 self._components.append(
230 ComponentEntry(
231 config_path=config_path,
232 var_name=var_name,
233 import_path=import_path,
234 args=args,
235 kwargs=kwargs,
236 )
237 )
239 def _analyze_func_entry(
240 self, config_path: str, var_name: str, value: dict[str, Any]
241 ) -> None:
242 """
243 Analyze a _func entry.
244 """
245 import_path = value["_func"]
246 module, name = _split_import_path(import_path)
247 self._imports[import_path] = (module, name)
248 component = ComponentEntry(
249 config_path=config_path,
250 var_name=var_name,
251 import_path=import_path,
252 is_func=True,
253 )
254 self._components.append(component)
256 def _process_arg_value(self, value: Any, config_path: str, deps: list[str]) -> Any:
257 """
258 Process a value that appears as an argument to a _type constructor.
260 Returns the value as-is for scalars, as a RefExpression for _ref,
261 or hoists nested _type entries to intermediate variables.
262 """
263 if not isinstance(value, dict):
264 if isinstance(value, list):
265 return [
266 self._process_arg_value(item, f"{config_path}.{idx}", deps)
267 for idx, item in enumerate(value)
268 ]
269 return value
271 if "_ref" in value:
272 ref_target = value["_ref"]
273 top = ref_target.split(".")[0]
274 deps.append(top)
275 return RefExpression(target=ref_target)
277 elif "_type" in value: 277 ↛ 282line 277 didn't jump to line 282 because the condition on line 277 was always true
278 anon_name = self._alloc_anon(config_path.replace(".", "_"))
279 self._analyze_type_entry(config_path, anon_name, value)
280 return RefExpression(target=anon_name)
282 elif "_func" in value:
283 anon_name = self._alloc_anon(config_path.replace(".", "_"))
284 self._analyze_func_entry(config_path, anon_name, value)
285 return RefExpression(target=anon_name)
287 elif value.get("_deep") is False:
288 return ...
290 elif "_entries" in value:
291 return self._analyze_entries(value["_entries"], config_path, deps)
293 else:
294 return value
296 def _analyze_entries(
297 self, entries: Any, config_path: str, deps: list[str]
298 ) -> EntriesExpression:
299 """
300 Analyze an _entries list.
301 """
302 items: list[tuple[Any, Any]] = list()
303 if not isinstance(entries, list): 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true
304 return EntriesExpression(items=[])
305 for idx, entry in enumerate(entries):
306 if not isinstance(entry, dict): 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true
307 continue
308 key = self._process_arg_value(entry.get("_key"), f"{config_path}.{idx}._key", deps)
309 value = self._process_arg_value(
310 entry.get("_value"), f"{config_path}.{idx}._value", deps
311 )
312 item = (key, value)
313 items.append(item)
314 return EntriesExpression(items=items)
316 def _topological_sort(self) -> list[str]:
317 """
318 Sort entries so that dependencies come before dependents.
319 """
320 visited: set[str] = set()
321 order: list[str] = list()
322 visiting: set[str] = set()
324 keys = {component.config_path for component in self._components}
325 keys.update({plain.config_path for plain in self._plains})
327 def visit(key_: str) -> None:
328 # What the absolute fuck?:w
329 if key_ in visited or key_ in visiting:
330 return
332 visiting.add(key_)
333 for dep in self._deps.get(key_, []):
334 if dep in keys: 334 ↛ 333line 334 didn't jump to line 333 because the condition on line 334 was always true
335 visit(dep)
336 visiting.discard(key_)
337 visited.add(key_)
338 order.append(key_)
340 for key in keys:
341 visit(key)
343 return order
346class CodeGenerator:
347 """
348 Generate Python source code from analysis results.
349 """
351 def __init__(self, analysis: AnalysisResult) -> None:
352 self._analysis = analysis
353 self._lines: list[str] = list()
354 self._mappings: list[SourceMapping] = list()
355 self._entry_map: dict[str, Any] = dict()
357 for component in analysis.components:
358 self._entry_map[component.config_path] = component
359 for plain in analysis.plains:
360 self._entry_map[plain.config_path] = plain
361 for component in analysis.components:
362 self._entry_map[component.var_name] = component
363 for plain in analysis.plains:
364 self._entry_map[plain.var_name] = plain
366 def generate(self) -> tuple[str, list[SourceMapping]]:
367 """
368 Generate the full source code.
369 """
370 self._emit_header()
371 self._emit_imports()
372 self._emit_blank()
373 self._emit_entries()
374 return "\n".join(self._lines) + "\n", list(self._mappings)
376 def _emit(self, line: str, config_path: str | None = None) -> None:
377 """
378 Emit a line of code, optionally recording its config path.
379 """
380 self._lines.append(line)
381 if config_path is not None:
382 mapping = SourceMapping(line=len(self._lines), config_path=config_path)
383 self._mappings.append(mapping)
385 def _emit_blank(self) -> None:
386 self._lines.append("")
388 def _emit_header(self) -> None:
389 self._emit("# Auto-generated by sygaldry for type checking — do not edit")
390 self._emit("from typing import Any")
392 def _emit_imports(self) -> None:
393 seen: set[str] = set()
394 for _, (mod, name) in sorted(self._analysis.imports.items(), key=lambda x: x[0]):
395 statement = f"from {mod} import {name}"
396 if statement not in seen: 396 ↛ 394line 396 didn't jump to line 394 because the condition on line 396 was always true
397 seen.add(statement)
398 self._emit(statement)
400 def _emit_entries(self) -> None:
401 """
402 Emit all entries in topological order.
403 """
404 emitted: set[str] = set()
405 for config_path in self._analysis.topological_order:
406 if config_path in emitted: 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true
407 continue
408 entry = self._entry_map.get(config_path)
409 if entry is None: 409 ↛ 410line 409 didn't jump to line 410 because the condition on line 409 was never true
410 continue
412 emitted.add(config_path)
414 if isinstance(entry, ComponentEntry):
415 self._emit_component(entry)
416 elif isinstance(entry, PlainEntry): 416 ↛ 405line 416 didn't jump to line 405 because the condition on line 416 was always true
417 self._emit_plain(entry)
419 def _emit_component(self, comp: ComponentEntry) -> None:
420 """
421 Emit a component (class instantiation or func import).
422 """
423 _, name = _split_import_path(comp.import_path)
424 if comp.is_func:
425 self._emit(
426 f"{comp.var_name} = {name} # config: {comp.config_path}",
427 config_path=comp.config_path,
428 )
429 return
431 parts = [self._value_to_source(arg) for arg in comp.args]
432 for key, value in comp.kwargs.items():
433 parts.append(f"{key}={self._value_to_source(value)}")
435 args_str = ", ".join(parts)
436 self._emit(
437 f"{comp.var_name}: {name} = {name}({args_str}) # config: {comp.config_path}",
438 config_path=comp.config_path,
439 )
441 def _emit_plain(self, plain: PlainEntry) -> None:
442 """
443 Emit a plain value assignment.
444 """
445 if isinstance(plain.value, RefExpression):
446 target = self._resolve_ref_source(plain.value.target)
447 self._emit(
448 f"{plain.var_name} = {target} # config: {plain.config_path}",
449 config_path=plain.config_path,
450 )
451 elif plain.value is ...:
452 self._emit(
453 f"{plain.var_name}: Any = ... # config: {plain.config_path}",
454 config_path=plain.config_path,
455 )
456 elif isinstance(plain.value, EntriesExpression):
457 items_str = ", ".join(
458 f"{self._value_to_source(key)}: {self._value_to_source(value)}"
459 for key, value in plain.value.items
460 )
461 self._emit(
462 f"{plain.var_name}: dict = {{{items_str}}} # config: {plain.config_path}",
463 config_path=plain.config_path,
464 )
465 else:
466 type_name = self._python_type_name(plain.value)
467 self._emit(
468 f"{plain.var_name}: {type_name} = {self._value_to_source(plain.value)}"
469 f" # config: {plain.config_path}",
470 config_path=plain.config_path,
471 )
473 def _resolve_ref_source(self, target: str) -> str:
474 """
475 Convert a ref target to a Python expression.
476 """
477 top, _, remainder = target.partition(".")
478 if entry := self._entry_map.get(top): 478 ↛ 484line 478 didn't jump to line 484 because the condition on line 478 was always true
479 if isinstance(entry, (ComponentEntry, PlainEntry)): 479 ↛ 482line 479 didn't jump to line 482 because the condition on line 479 was always true
480 var_name = entry.var_name
481 else:
482 var_name = _to_identifier(top)
483 else:
484 var_name = _to_identifier(top)
485 if remainder:
486 return f"{var_name}.{remainder}"
487 return var_name
489 def _value_to_source(self, value: Any) -> str:
490 """
491 Convert a config value to its Python source representation.
492 """
493 if isinstance(value, RefExpression):
494 return self._resolve_ref_source(value.target)
495 elif isinstance(value, EntriesExpression): 495 ↛ 496line 495 didn't jump to line 496 because the condition on line 495 was never true
496 items_str = ", ".join(
497 f"{self._value_to_source(k)}: {self._value_to_source(v)}"
498 for k, v in value.items
499 )
500 return f"{{{items_str}}}"
501 elif value is ...: 501 ↛ 502line 501 didn't jump to line 502 because the condition on line 501 was never true
502 return "..."
503 elif value is None:
504 return "None"
505 elif isinstance(value, bool):
506 return "True" if value else "False"
507 elif isinstance(value, (float, int, str)):
508 return repr(value)
509 elif isinstance(value, list):
510 items = ", ".join(self._value_to_source(v) for v in value)
511 return f"[{items}]"
512 elif isinstance(value, dict): 512 ↛ 519line 512 didn't jump to line 519 because the condition on line 512 was always true
513 items = ", ".join(
514 f"{self._value_to_source(k)}: {self._value_to_source(v)}"
515 for k, v in value.items()
516 )
517 return f"{{{items}}}"
518 else:
519 return repr(value)
521 @staticmethod
522 def _python_type_name(value: Any) -> str:
523 """
524 Get a type annotation name for a plain value.
525 """
526 if value is None:
527 return "None"
528 elif isinstance(value, bool):
529 return "bool"
530 elif isinstance(value, int):
531 return "int"
532 elif isinstance(value, float):
533 return "float"
534 elif isinstance(value, str):
535 return "str"
536 elif isinstance(value, list):
537 return "list"
538 elif isinstance(value, dict): 538 ↛ 541line 538 didn't jump to line 541 because the condition on line 538 was always true
539 return "dict"
540 else:
541 return "Any"
544def generate_check_source(
545 config: dict[str, Any],
546) -> tuple[str, list[SourceMapping]]:
547 """
548 Generate type-checkable Python source from a sygaldry config.
550 :param config: Loaded, interpolated config mapping.
551 :type config: dict
552 :returns: ``(source_code, mappings)`` tuple.
553 """
554 analyzer = ConfigAnalyzer(config)
555 result = analyzer.analyze()
556 generator = CodeGenerator(result)
557 return generator.generate()