Coverage for sygaldry / codegen.py: 84%
369 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-15 18:36 -0400
1from __future__ import annotations
3__author__ = "Rohan B. Dalton"
5import importlib
6import inspect
7import re
8from dataclasses import dataclass, field
9from typing import Any, Optional
11from .types import RESERVED_KEYS
13_NON_IDENTIFIER_RE = re.compile(r"[^a-zA-Z0-9_]")
16@dataclass(frozen=True)
17class SourceMapping:
18 """
19 Map a generated line number to a config path.
20 """
22 line: int
23 config_path: str
26@dataclass
27class ComponentEntry:
28 """
29 A typed component extracted from the config.
30 """
32 config_path: str
33 var_name: str
34 import_path: str
35 args: list[Any] = field(default_factory=list)
36 kwargs: dict[str, Any] = field(default_factory=dict)
37 is_func: bool = False
40@dataclass
41class PlainEntry:
42 """
43 A plain (non-typed) config value.
44 """
46 config_path: str
47 var_name: str
48 value: Any
51@dataclass
52class RefExpression:
53 """
54 A reference expression to be emitted as a variable or attribute access.
55 """
57 target: str
60@dataclass
61class EntriesExpression:
62 """
63 A dict built from _entries.
64 """
66 items: list[tuple[Any, Any]]
69@dataclass
70class AnalysisResult:
71 """
72 Result of analyzing a config dict.
73 """
75 components: list[ComponentEntry] = field(default_factory=list)
76 plains: list[PlainEntry] = field(default_factory=list)
77 topological_order: list[str] = field(default_factory=list)
78 imports: dict[str, tuple[str, str]] = field(default_factory=dict)
81def _to_identifier(key: str) -> str:
82 """
83 Convert a config key to a valid Python identifier.
84 """
85 if result := _NON_IDENTIFIER_RE.sub("_", key):
86 if result[0].isdigit():
87 result = f"_{result}"
88 return result
89 else:
90 return "_unnamed"
93def _split_import_path(dotted: str) -> tuple[str, str]:
94 """
95 Split 'a.b.ClassName' into ('a.b', 'ClassName').
96 """
97 parts = dotted.rsplit(".", 1)
98 if len(parts) == 1:
99 return parts[0], parts[0]
100 else:
101 return parts[0], parts[1]
104class ConfigAnalyzer:
105 """
106 Walk a loaded config dict and extract typed component specs.
107 """
109 def __init__(self, config: dict[str, Any]) -> None:
110 self._config = config
111 self._components: list[ComponentEntry] = list()
112 self._plains: list[PlainEntry] = list()
113 self._imports: dict[str, tuple[str, str]] = dict()
114 self._deps: dict[str, list[str]] = dict()
115 self._variable_names: dict[str, str] = dict()
116 self._anon_counter = 0
117 self._used_names: set[str] = set()
119 def analyze(self) -> AnalysisResult:
120 """
121 Analyze the config and return structured results.
122 """
123 for key, value in self._config.items():
124 self._analyze_top_level(key, value)
125 order = self._topological_sort()
126 result = AnalysisResult(
127 components=self._components,
128 plains=self._plains,
129 topological_order=order,
130 imports=self._imports,
131 )
132 return result
134 def _alloc_name(self, preferred: str) -> str:
135 """
136 Allocate a unique variable name.
137 """
138 name = _to_identifier(preferred)
139 if name not in self._used_names: 139 ↛ 143line 139 didn't jump to line 143 because the condition on line 139 was always true
140 self._used_names.add(name)
141 return name
142 else:
143 counter = 0
144 while f"{name}_{counter}" in self._used_names:
145 counter += 1
146 unique = f"{name}_{counter}"
147 self._used_names.add(unique)
148 return unique
150 def _alloc_anon(self, hint: str) -> str:
151 """
152 Allocate an anonymous variable name.
153 """
154 name = f"__sygaldry_{_to_identifier(hint)}_{self._anon_counter}"
155 self._anon_counter += 1
156 self._used_names.add(name)
157 return name
159 def _analyze_top_level(self, key: str, value: Any) -> None:
160 """
161 Analyze a single top-level config entry.
162 """
163 variable_name = self._alloc_name(key)
164 self._variable_names[key] = variable_name
165 self._deps[key] = list()
167 if isinstance(value, dict):
168 if "_ref" in value:
169 ref_target = value["_ref"]
170 top = ref_target.split(".")[0]
171 self._deps[key].append(top)
172 self._plains.append(
173 PlainEntry(
174 config_path=key,
175 var_name=variable_name,
176 value=RefExpression(target=ref_target),
177 )
178 )
179 elif "_type" in value:
180 self._analyze_type_entry(key, variable_name, value)
181 elif "_func" in value:
182 self._analyze_func_entry(key, variable_name, value)
183 elif value.get("_deep") is False:
184 self._plains.append(
185 PlainEntry(config_path=key, var_name=variable_name, value=...)
186 )
187 elif "_entries" in value:
188 entries_expr = self._analyze_entries(value["_entries"], key, self._deps[key])
189 self._plains.append(
190 PlainEntry(config_path=key, var_name=variable_name, value=entries_expr)
191 )
192 else:
193 self._plains.append(
194 PlainEntry(config_path=key, var_name=variable_name, value=value)
195 )
196 else:
197 self._plains.append(
198 PlainEntry(config_path=key, var_name=variable_name, value=value)
199 )
201 def _analyze_type_entry(
202 self, config_path: str, var_name: str, entry: dict[str, Any]
203 ) -> None:
204 """
205 Analyze a _type entry.
206 """
207 import_path = entry["_type"]
208 mod, name = _split_import_path(import_path)
209 self._imports[import_path] = (mod, name)
211 raw_args = entry.get("_args", [])
212 args = list()
213 for idx, arg in enumerate(raw_args):
214 processed = self._process_arg_value(
215 arg, f"{config_path}._args.{idx}", self._deps.setdefault(config_path, [])
216 )
217 args.append(processed)
219 kwargs: dict[str, Any] = dict()
220 for key, value in entry.items():
221 if key in RESERVED_KEYS:
222 continue
223 processed = self._process_arg_value(
224 value, f"{config_path}.{key}", self._deps.setdefault(config_path, [])
225 )
226 kwargs[key] = processed
228 raw_kwargs = entry.get("_kwargs", {})
229 for key, value in raw_kwargs.items():
230 processed = self._process_arg_value(
231 value, f"{config_path}._kwargs.{key}", self._deps.setdefault(config_path, [])
232 )
233 kwargs[key] = processed
235 self._components.append(
236 ComponentEntry(
237 config_path=config_path,
238 var_name=var_name,
239 import_path=import_path,
240 args=args,
241 kwargs=kwargs,
242 )
243 )
245 def _analyze_func_entry(
246 self, config_path: str, var_name: str, value: dict[str, Any]
247 ) -> None:
248 """
249 Analyze a _func entry.
250 """
251 import_path = value["_func"]
252 module, name = _split_import_path(import_path)
253 self._imports[import_path] = (module, name)
254 component = ComponentEntry(
255 config_path=config_path,
256 var_name=var_name,
257 import_path=import_path,
258 is_func=True,
259 )
260 self._components.append(component)
262 def _process_arg_value(self, value: Any, config_path: str, deps: list[str]) -> Any:
263 """
264 Process a value that appears as an argument to a _type constructor.
266 Returns the value as-is for scalars, as a RefExpression for _ref,
267 or hoists nested _type entries to intermediate variables.
268 """
269 if not isinstance(value, dict):
270 if isinstance(value, list):
271 return [
272 self._process_arg_value(item, f"{config_path}.{idx}", deps)
273 for idx, item in enumerate(value)
274 ]
275 return value
277 if "_ref" in value:
278 ref_target = value["_ref"]
279 top = ref_target.split(".")[0]
280 deps.append(top)
281 return RefExpression(target=ref_target)
283 elif "_type" in value: 283 ↛ 288line 283 didn't jump to line 288 because the condition on line 283 was always true
284 anon_name = self._alloc_anon(config_path.replace(".", "_"))
285 self._analyze_type_entry(config_path, anon_name, value)
286 return RefExpression(target=anon_name)
288 elif "_func" in value:
289 anon_name = self._alloc_anon(config_path.replace(".", "_"))
290 self._analyze_func_entry(config_path, anon_name, value)
291 return RefExpression(target=anon_name)
293 elif value.get("_deep") is False:
294 return ...
296 elif "_entries" in value:
297 return self._analyze_entries(value["_entries"], config_path, deps)
299 else:
300 return value
302 def _analyze_entries(
303 self, entries: Any, config_path: str, deps: list[str]
304 ) -> EntriesExpression:
305 """
306 Analyze an _entries list.
307 """
308 items: list[tuple[Any, Any]] = list()
309 if not isinstance(entries, list): 309 ↛ 310line 309 didn't jump to line 310 because the condition on line 309 was never true
310 return EntriesExpression(items=[])
311 for idx, entry in enumerate(entries):
312 if not isinstance(entry, dict): 312 ↛ 313line 312 didn't jump to line 313 because the condition on line 312 was never true
313 continue
314 key = self._process_arg_value(entry.get("_key"), f"{config_path}.{idx}._key", deps)
315 value = self._process_arg_value(
316 entry.get("_value"), f"{config_path}.{idx}._value", deps
317 )
318 item = (key, value)
319 items.append(item)
320 return EntriesExpression(items=items)
322 def _topological_sort(self) -> list[str]:
323 """
324 Sort entries so that dependencies come before dependents.
325 """
326 visited: set[str] = set()
327 order: list[str] = list()
328 visiting: set[str] = set()
330 keys = {component.config_path for component in self._components}
331 keys.update({plain.config_path for plain in self._plains})
333 def visit(key_: str) -> None:
334 # What the absolute fuck?:w
335 if key_ in visited or key_ in visiting:
336 return
338 visiting.add(key_)
339 for dep in self._deps.get(key_, []):
340 if dep in keys: 340 ↛ 339line 340 didn't jump to line 339 because the condition on line 340 was always true
341 visit(dep)
342 visiting.discard(key_)
343 visited.add(key_)
344 order.append(key_)
346 for key in keys:
347 visit(key)
349 return order
352class CodeGenerator:
353 """
354 Generate Python source code from analysis results.
355 """
357 def __init__(self, analysis: AnalysisResult, *, line_length: int = 99) -> None:
358 self._analysis = analysis
359 self._line_length = line_length
360 self._lines: list[str] = list()
361 self._mappings: list[SourceMapping] = list()
362 self._entry_map: dict[str, Any] = dict()
364 for component in analysis.components:
365 self._entry_map[component.config_path] = component
366 for plain in analysis.plains:
367 self._entry_map[plain.config_path] = plain
368 for component in analysis.components:
369 self._entry_map[component.var_name] = component
370 for plain in analysis.plains:
371 self._entry_map[plain.var_name] = plain
373 def generate(
374 self,
375 call_target: str | None = None,
376 call_method: str | None = None,
377 call_args: list[Any] | None = None,
378 ) -> tuple[str, list[SourceMapping]]:
379 """
380 Generate the full source code.
381 """
382 needs_asyncio = call_target is not None and self._is_async_call(
383 call_target, call_method
384 )
385 self._emit_header(needs_asyncio=needs_asyncio)
386 self._emit_imports()
387 self._emit_blank()
388 self._emit_entries()
389 if call_target is not None:
390 self._emit_blank()
391 self._emit_call(call_target, call_method, call_args or [])
392 return "\n".join(self._lines) + "\n", list(self._mappings)
394 def _emit(self, line: str, config_path: str | None = None) -> None:
395 """
396 Emit a line of code, optionally recording its config path.
397 """
398 self._lines.append(line)
399 if config_path is not None:
400 mapping = SourceMapping(line=len(self._lines), config_path=config_path)
401 self._mappings.append(mapping)
403 def _emit_blank(self) -> None:
404 self._lines.append("")
406 def _emit_header(self, *, needs_asyncio: bool = False) -> None:
407 self._emit("# Auto-generated by sygaldry for type checking — do not edit")
408 if needs_asyncio:
409 self._emit("import asyncio")
410 self._emit("from typing import Any")
412 def _emit_imports(self) -> None:
413 seen: set[str] = set()
414 for _, (mod, name) in sorted(self._analysis.imports.items(), key=lambda x: x[0]):
415 statement = f"from {mod} import {name}"
416 if statement not in seen: 416 ↛ 414line 416 didn't jump to line 414 because the condition on line 416 was always true
417 seen.add(statement)
418 self._emit(statement)
420 def _emit_entries(self) -> None:
421 """
422 Emit all entries in topological order.
423 """
424 emitted: set[str] = set()
425 for config_path in self._analysis.topological_order:
426 if config_path in emitted: 426 ↛ 427line 426 didn't jump to line 427 because the condition on line 426 was never true
427 continue
428 entry = self._entry_map.get(config_path)
429 if entry is None: 429 ↛ 430line 429 didn't jump to line 430 because the condition on line 429 was never true
430 continue
432 emitted.add(config_path)
434 if isinstance(entry, ComponentEntry):
435 self._emit_component(entry)
436 elif isinstance(entry, PlainEntry): 436 ↛ 425line 436 didn't jump to line 425 because the condition on line 436 was always true
437 self._emit_plain(entry)
439 def _is_async_call(self, target: str, method: str | None) -> bool:
440 """
441 Check whether the call target's method (or __call__) is async.
442 """
443 entry = self._entry_map.get(target)
444 if not isinstance(entry, ComponentEntry): 444 ↛ 445line 444 didn't jump to line 445 because the condition on line 444 was never true
445 return False
446 try:
447 mod_path, cls_name = _split_import_path(entry.import_path)
448 mod = importlib.import_module(mod_path)
449 cls = getattr(mod, cls_name)
450 if method:
451 func = getattr(cls, method, None)
452 else:
453 func = getattr(cls, "__call__", None)
454 return inspect.iscoroutinefunction(func)
455 except Exception:
456 return False
458 def _emit_call(
459 self,
460 target: str,
461 method: str | None,
462 args: list[Any],
463 ) -> None:
464 """
465 Emit a call expression at the end of the generated code.
466 """
467 var_name = _to_identifier(target)
468 if entry := self._entry_map.get(target): 468 ↛ 472line 468 didn't jump to line 472 because the condition on line 468 was always true
469 if isinstance(entry, (ComponentEntry, PlainEntry)): 469 ↛ 472line 469 didn't jump to line 472 because the condition on line 469 was always true
470 var_name = entry.var_name
472 is_async = self._is_async_call(target, method)
474 callee = f"{var_name}.{method}" if method else var_name
475 arg_parts = [self._value_to_source(arg) for arg in args]
476 args_str = ", ".join(arg_parts)
477 call_expr = f"{callee}({args_str})"
479 if is_async:
480 one_line = f"asyncio.run({call_expr})"
481 if len(one_line) <= self._line_length or not arg_parts: 481 ↛ 484line 481 didn't jump to line 484 because the condition on line 481 was always true
482 self._emit(one_line)
483 else:
484 self._emit(f"asyncio.run({callee}(")
485 for idx, arg_part in enumerate(arg_parts):
486 suffix = "," if idx < len(arg_parts) - 1 else ""
487 self._emit(f" {arg_part}{suffix}")
488 self._emit("))")
489 elif len(call_expr) <= self._line_length or not arg_parts: 489 ↛ 492line 489 didn't jump to line 492 because the condition on line 489 was always true
490 self._emit(call_expr)
491 else:
492 self._emit(f"{callee}(")
493 for idx, arg_part in enumerate(arg_parts):
494 suffix = "," if idx < len(arg_parts) - 1 else ""
495 self._emit(f" {arg_part}{suffix}")
496 self._emit(")")
498 def _emit_component(self, comp: ComponentEntry) -> None:
499 """
500 Emit a component (class instantiation or func import).
501 """
502 _, name = _split_import_path(comp.import_path)
503 if comp.is_func:
504 self._emit(
505 f"{comp.var_name} = {name} # config: {comp.config_path}",
506 config_path=comp.config_path,
507 )
508 return
510 parts = [self._value_to_source(arg) for arg in comp.args]
511 for key, value in comp.kwargs.items():
512 parts.append(f"{key}={self._value_to_source(value)}")
514 comment = f" # config: {comp.config_path}"
515 args_str = ", ".join(parts)
516 one_line = f"{comp.var_name}: {name} = {name}({args_str}){comment}"
517 if len(one_line) <= self._line_length or not parts: 517 ↛ 520line 517 didn't jump to line 520 because the condition on line 517 was always true
518 self._emit(one_line, config_path=comp.config_path)
519 else:
520 self._emit(
521 f"{comp.var_name}: {name} = {name}({comment}", config_path=comp.config_path
522 )
523 for part in parts:
524 self._emit(f" {part},")
525 self._emit(")")
527 def _emit_plain(self, plain: PlainEntry) -> None:
528 """
529 Emit a plain value assignment.
530 """
531 comment = f" # config: {plain.config_path}"
532 if isinstance(plain.value, RefExpression):
533 target = self._resolve_ref_source(plain.value.target)
534 self._emit(
535 f"{plain.var_name} = {target}{comment}",
536 config_path=plain.config_path,
537 )
538 elif plain.value is ...:
539 self._emit(
540 f"{plain.var_name}: Any = ...{comment}",
541 config_path=plain.config_path,
542 )
543 elif isinstance(plain.value, EntriesExpression):
544 items = [
545 (self._value_to_source(key), self._value_to_source(value))
546 for key, value in plain.value.items
547 ]
548 items_str = ", ".join(f"{k}: {v}" for k, v in items)
549 one_line = f"{plain.var_name}: dict = {{{items_str}}}{comment}"
550 if len(one_line) <= self._line_length or not items: 550 ↛ 553line 550 didn't jump to line 553 because the condition on line 550 was always true
551 self._emit(one_line, config_path=plain.config_path)
552 else:
553 self._emit(
554 f"{plain.var_name}: dict = {{{comment}", config_path=plain.config_path
555 )
556 for i, (k, v) in enumerate(items):
557 suffix = ","
558 self._emit(f" {k}: {v}{suffix}")
559 self._emit("}")
560 else:
561 type_name = self._python_type_name(plain.value)
562 self._emit(
563 f"{plain.var_name}: {type_name} = {self._value_to_source(plain.value)}{comment}",
564 config_path=plain.config_path,
565 )
567 def _resolve_ref_source(self, target: str) -> str:
568 """
569 Convert a ref target to a Python expression.
570 """
571 top, _, remainder = target.partition(".")
572 if entry := self._entry_map.get(top): 572 ↛ 578line 572 didn't jump to line 578 because the condition on line 572 was always true
573 if isinstance(entry, (ComponentEntry, PlainEntry)): 573 ↛ 576line 573 didn't jump to line 576 because the condition on line 573 was always true
574 var_name = entry.var_name
575 else:
576 var_name = _to_identifier(top)
577 else:
578 var_name = _to_identifier(top)
579 if remainder:
580 return f"{var_name}.{remainder}"
581 return var_name
583 def _value_to_source(self, value: Any) -> str:
584 """
585 Convert a config value to its Python source representation.
586 """
587 if isinstance(value, RefExpression):
588 return self._resolve_ref_source(value.target)
589 elif isinstance(value, EntriesExpression): 589 ↛ 590line 589 didn't jump to line 590 because the condition on line 589 was never true
590 items_str = ", ".join(
591 f"{self._value_to_source(k)}: {self._value_to_source(v)}"
592 for k, v in value.items
593 )
594 return f"{{{items_str}}}"
595 elif value is ...: 595 ↛ 596line 595 didn't jump to line 596 because the condition on line 595 was never true
596 return "..."
597 elif value is None:
598 return "None"
599 elif isinstance(value, bool):
600 return "True" if value else "False"
601 elif isinstance(value, (float, int, str)):
602 return repr(value)
603 elif isinstance(value, list):
604 items = ", ".join(self._value_to_source(v) for v in value)
605 return f"[{items}]"
606 elif isinstance(value, dict): 606 ↛ 613line 606 didn't jump to line 613 because the condition on line 606 was always true
607 items = ", ".join(
608 f"{self._value_to_source(k)}: {self._value_to_source(v)}"
609 for k, v in value.items()
610 )
611 return f"{{{items}}}"
612 else:
613 return repr(value)
615 @staticmethod
616 def _python_type_name(value: Any) -> str:
617 """
618 Get a type annotation name for a plain value.
619 """
620 if value is None:
621 return "None"
622 elif isinstance(value, bool):
623 return "bool"
624 elif isinstance(value, int):
625 return "int"
626 elif isinstance(value, float):
627 return "float"
628 elif isinstance(value, str):
629 return "str"
630 elif isinstance(value, list):
631 return "list"
632 elif isinstance(value, dict): 632 ↛ 635line 632 didn't jump to line 635 because the condition on line 632 was always true
633 return "dict"
634 else:
635 return "Any"
638def generate_check_source(
639 config: dict[str, Any],
640 *,
641 line_length: int = 99,
642 call_target: str | None = None,
643 call_method: str | None = None,
644 call_args: list[Any] | None = None,
645) -> tuple[str, list[SourceMapping]]:
646 """
647 Generate type-checkable Python source from a sygaldry config.
649 :param config: Loaded, interpolated config mapping.
650 :param line_length: Maximum line length for generated code.
651 :param call_target: Top-level config key to emit a call expression for.
652 :param call_method: Method name to call on the target.
653 :param call_args: Arguments for the call expression.
654 :type config: dict
655 :type line_length: int
656 :returns: ``(source_code, mappings)`` tuple.
657 """
658 analyzer = ConfigAnalyzer(config)
659 result = analyzer.analyze()
660 generator = CodeGenerator(result, line_length=line_length)
661 return generator.generate(
662 call_target=call_target,
663 call_method=call_method,
664 call_args=call_args,
665 )