Coverage for sygaldry / codegen.py: 89%
304 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-06 19:06 -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(
183 PlainEntry(config_path=key, var_name=variable_name, value=...)
184 )
185 elif "_entries" in value:
186 entries_expr = self._analyze_entries(value["_entries"], key, self._deps[key])
187 self._plains.append(
188 PlainEntry(config_path=key, var_name=variable_name, value=entries_expr)
189 )
190 else:
191 self._plains.append(
192 PlainEntry(config_path=key, var_name=variable_name, value=value)
193 )
194 else:
195 self._plains.append(
196 PlainEntry(config_path=key, var_name=variable_name, value=value)
197 )
199 def _analyze_type_entry(
200 self, config_path: str, var_name: str, entry: dict[str, Any]
201 ) -> None:
202 """
203 Analyze a _type entry.
204 """
205 import_path = entry["_type"]
206 mod, name = _split_import_path(import_path)
207 self._imports[import_path] = (mod, name)
209 raw_args = entry.get("_args", [])
210 args = list()
211 for idx, arg in enumerate(raw_args):
212 processed = self._process_arg_value(
213 arg, f"{config_path}._args.{idx}", self._deps.setdefault(config_path, [])
214 )
215 args.append(processed)
217 kwargs: dict[str, Any] = dict()
218 for key, value in entry.items():
219 if key in RESERVED_KEYS:
220 continue
221 processed = self._process_arg_value(
222 value, f"{config_path}.{key}", self._deps.setdefault(config_path, [])
223 )
224 kwargs[key] = processed
226 raw_kwargs = entry.get("_kwargs", {})
227 for key, value in raw_kwargs.items():
228 processed = self._process_arg_value(
229 value, f"{config_path}._kwargs.{key}", self._deps.setdefault(config_path, [])
230 )
231 kwargs[key] = processed
233 self._components.append(
234 ComponentEntry(
235 config_path=config_path,
236 var_name=var_name,
237 import_path=import_path,
238 args=args,
239 kwargs=kwargs,
240 )
241 )
243 def _analyze_func_entry(
244 self, config_path: str, var_name: str, value: dict[str, Any]
245 ) -> None:
246 """
247 Analyze a _func entry.
248 """
249 import_path = value["_func"]
250 module, name = _split_import_path(import_path)
251 self._imports[import_path] = (module, name)
252 component = ComponentEntry(
253 config_path=config_path,
254 var_name=var_name,
255 import_path=import_path,
256 is_func=True,
257 )
258 self._components.append(component)
260 def _process_arg_value(self, value: Any, config_path: str, deps: list[str]) -> Any:
261 """
262 Process a value that appears as an argument to a _type constructor.
264 Returns the value as-is for scalars, as a RefExpression for _ref,
265 or hoists nested _type entries to intermediate variables.
266 """
267 if not isinstance(value, dict):
268 if isinstance(value, list):
269 return [
270 self._process_arg_value(item, f"{config_path}.{idx}", deps)
271 for idx, item in enumerate(value)
272 ]
273 return value
275 if "_ref" in value:
276 ref_target = value["_ref"]
277 top = ref_target.split(".")[0]
278 deps.append(top)
279 return RefExpression(target=ref_target)
281 elif "_type" in value: 281 ↛ 286line 281 didn't jump to line 286 because the condition on line 281 was always true
282 anon_name = self._alloc_anon(config_path.replace(".", "_"))
283 self._analyze_type_entry(config_path, anon_name, value)
284 return RefExpression(target=anon_name)
286 elif "_func" in value:
287 anon_name = self._alloc_anon(config_path.replace(".", "_"))
288 self._analyze_func_entry(config_path, anon_name, value)
289 return RefExpression(target=anon_name)
291 elif value.get("_deep") is False:
292 return ...
294 elif "_entries" in value:
295 return self._analyze_entries(value["_entries"], config_path, deps)
297 else:
298 return value
300 def _analyze_entries(
301 self, entries: Any, config_path: str, deps: list[str]
302 ) -> EntriesExpression:
303 """
304 Analyze an _entries list.
305 """
306 items: list[tuple[Any, Any]] = list()
307 if not isinstance(entries, list): 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true
308 return EntriesExpression(items=[])
309 for idx, entry in enumerate(entries):
310 if not isinstance(entry, dict): 310 ↛ 311line 310 didn't jump to line 311 because the condition on line 310 was never true
311 continue
312 key = self._process_arg_value(entry.get("_key"), f"{config_path}.{idx}._key", deps)
313 value = self._process_arg_value(
314 entry.get("_value"), f"{config_path}.{idx}._value", deps
315 )
316 item = (key, value)
317 items.append(item)
318 return EntriesExpression(items=items)
320 def _topological_sort(self) -> list[str]:
321 """
322 Sort entries so that dependencies come before dependents.
323 """
324 visited: set[str] = set()
325 order: list[str] = list()
326 visiting: set[str] = set()
328 keys = {component.config_path for component in self._components}
329 keys.update({plain.config_path for plain in self._plains})
331 def visit(key_: str) -> None:
332 # What the absolute fuck?:w
333 if key_ in visited or key_ in visiting:
334 return
336 visiting.add(key_)
337 for dep in self._deps.get(key_, []):
338 if dep in keys: 338 ↛ 337line 338 didn't jump to line 337 because the condition on line 338 was always true
339 visit(dep)
340 visiting.discard(key_)
341 visited.add(key_)
342 order.append(key_)
344 for key in keys:
345 visit(key)
347 return order
350class CodeGenerator:
351 """
352 Generate Python source code from analysis results.
353 """
355 def __init__(self, analysis: AnalysisResult) -> None:
356 self._analysis = analysis
357 self._lines: list[str] = list()
358 self._mappings: list[SourceMapping] = list()
359 self._entry_map: dict[str, Any] = dict()
361 for component in analysis.components:
362 self._entry_map[component.config_path] = component
363 for plain in analysis.plains:
364 self._entry_map[plain.config_path] = plain
365 for component in analysis.components:
366 self._entry_map[component.var_name] = component
367 for plain in analysis.plains:
368 self._entry_map[plain.var_name] = plain
370 def generate(self) -> tuple[str, list[SourceMapping]]:
371 """
372 Generate the full source code.
373 """
374 self._emit_header()
375 self._emit_imports()
376 self._emit_blank()
377 self._emit_entries()
378 return "\n".join(self._lines) + "\n", list(self._mappings)
380 def _emit(self, line: str, config_path: str | None = None) -> None:
381 """
382 Emit a line of code, optionally recording its config path.
383 """
384 self._lines.append(line)
385 if config_path is not None:
386 mapping = SourceMapping(line=len(self._lines), config_path=config_path)
387 self._mappings.append(mapping)
389 def _emit_blank(self) -> None:
390 self._lines.append("")
392 def _emit_header(self) -> None:
393 self._emit("# Auto-generated by sygaldry for type checking — do not edit")
394 self._emit("from typing import Any")
396 def _emit_imports(self) -> None:
397 seen: set[str] = set()
398 for _, (mod, name) in sorted(self._analysis.imports.items(), key=lambda x: x[0]):
399 statement = f"from {mod} import {name}"
400 if statement not in seen: 400 ↛ 398line 400 didn't jump to line 398 because the condition on line 400 was always true
401 seen.add(statement)
402 self._emit(statement)
404 def _emit_entries(self) -> None:
405 """
406 Emit all entries in topological order.
407 """
408 emitted: set[str] = set()
409 for config_path in self._analysis.topological_order:
410 if config_path in emitted: 410 ↛ 411line 410 didn't jump to line 411 because the condition on line 410 was never true
411 continue
412 entry = self._entry_map.get(config_path)
413 if entry is None: 413 ↛ 414line 413 didn't jump to line 414 because the condition on line 413 was never true
414 continue
416 emitted.add(config_path)
418 if isinstance(entry, ComponentEntry):
419 self._emit_component(entry)
420 elif isinstance(entry, PlainEntry): 420 ↛ 409line 420 didn't jump to line 409 because the condition on line 420 was always true
421 self._emit_plain(entry)
423 def _emit_component(self, comp: ComponentEntry) -> None:
424 """
425 Emit a component (class instantiation or func import).
426 """
427 _, name = _split_import_path(comp.import_path)
428 if comp.is_func:
429 self._emit(
430 f"{comp.var_name} = {name} # config: {comp.config_path}",
431 config_path=comp.config_path,
432 )
433 return
435 parts = [self._value_to_source(arg) for arg in comp.args]
436 for key, value in comp.kwargs.items():
437 parts.append(f"{key}={self._value_to_source(value)}")
439 args_str = ", ".join(parts)
440 self._emit(
441 f"{comp.var_name}: {name} = {name}({args_str}) # config: {comp.config_path}",
442 config_path=comp.config_path,
443 )
445 def _emit_plain(self, plain: PlainEntry) -> None:
446 """
447 Emit a plain value assignment.
448 """
449 if isinstance(plain.value, RefExpression):
450 target = self._resolve_ref_source(plain.value.target)
451 self._emit(
452 f"{plain.var_name} = {target} # config: {plain.config_path}",
453 config_path=plain.config_path,
454 )
455 elif plain.value is ...:
456 self._emit(
457 f"{plain.var_name}: Any = ... # config: {plain.config_path}",
458 config_path=plain.config_path,
459 )
460 elif isinstance(plain.value, EntriesExpression):
461 items_str = ", ".join(
462 f"{self._value_to_source(key)}: {self._value_to_source(value)}"
463 for key, value in plain.value.items
464 )
465 self._emit(
466 f"{plain.var_name}: dict = {{{items_str}}} # config: {plain.config_path}",
467 config_path=plain.config_path,
468 )
469 else:
470 type_name = self._python_type_name(plain.value)
471 self._emit(
472 f"{plain.var_name}: {type_name} = {self._value_to_source(plain.value)}"
473 f" # config: {plain.config_path}",
474 config_path=plain.config_path,
475 )
477 def _resolve_ref_source(self, target: str) -> str:
478 """
479 Convert a ref target to a Python expression.
480 """
481 top, _, remainder = target.partition(".")
482 if entry := self._entry_map.get(top): 482 ↛ 488line 482 didn't jump to line 488 because the condition on line 482 was always true
483 if isinstance(entry, (ComponentEntry, PlainEntry)): 483 ↛ 486line 483 didn't jump to line 486 because the condition on line 483 was always true
484 var_name = entry.var_name
485 else:
486 var_name = _to_identifier(top)
487 else:
488 var_name = _to_identifier(top)
489 if remainder:
490 return f"{var_name}.{remainder}"
491 return var_name
493 def _value_to_source(self, value: Any) -> str:
494 """
495 Convert a config value to its Python source representation.
496 """
497 if isinstance(value, RefExpression):
498 return self._resolve_ref_source(value.target)
499 elif isinstance(value, EntriesExpression): 499 ↛ 500line 499 didn't jump to line 500 because the condition on line 499 was never true
500 items_str = ", ".join(
501 f"{self._value_to_source(k)}: {self._value_to_source(v)}"
502 for k, v in value.items
503 )
504 return f"{{{items_str}}}"
505 elif value is ...: 505 ↛ 506line 505 didn't jump to line 506 because the condition on line 505 was never true
506 return "..."
507 elif value is None:
508 return "None"
509 elif isinstance(value, bool):
510 return "True" if value else "False"
511 elif isinstance(value, (float, int, str)):
512 return repr(value)
513 elif isinstance(value, list):
514 items = ", ".join(self._value_to_source(v) for v in value)
515 return f"[{items}]"
516 elif isinstance(value, dict): 516 ↛ 523line 516 didn't jump to line 523 because the condition on line 516 was always true
517 items = ", ".join(
518 f"{self._value_to_source(k)}: {self._value_to_source(v)}"
519 for k, v in value.items()
520 )
521 return f"{{{items}}}"
522 else:
523 return repr(value)
525 @staticmethod
526 def _python_type_name(value: Any) -> str:
527 """
528 Get a type annotation name for a plain value.
529 """
530 if value is None:
531 return "None"
532 elif isinstance(value, bool):
533 return "bool"
534 elif isinstance(value, int):
535 return "int"
536 elif isinstance(value, float):
537 return "float"
538 elif isinstance(value, str):
539 return "str"
540 elif isinstance(value, list):
541 return "list"
542 elif isinstance(value, dict): 542 ↛ 545line 542 didn't jump to line 545 because the condition on line 542 was always true
543 return "dict"
544 else:
545 return "Any"
548def generate_check_source(
549 config: dict[str, Any],
550) -> tuple[str, list[SourceMapping]]:
551 """
552 Generate type-checkable Python source from a sygaldry config.
554 :param config: Loaded, interpolated config mapping.
555 :type config: dict
556 :returns: ``(source_code, mappings)`` tuple.
557 """
558 analyzer = ConfigAnalyzer(config)
559 result = analyzer.analyze()
560 generator = CodeGenerator(result)
561 return generator.generate()