from dataclasses import dataclass, field
from typing import Any, cast
from pydantic import BaseModel
from codegen.operation import (
EmptyOptions,
OperationEntry,
OperationRegistry,
)
from codegen.scope import PROJECT, Scope, ScopeTree, discover_scopes
from codegen.store import BuildStore, NodePath
[docs]
@dataclass
class BuildContext[InstanceT, ConfigT: BaseModel]:
config: ConfigT
scope: Scope
instance: InstanceT
instance_id: NodePath
store: BuildStore
package_prefix: str = ""
[docs]
@dataclass
class Engine:
registry: OperationRegistry = field(default_factory=OperationRegistry)
package_prefix: str = ""
def build(self, config: BaseModel) -> BuildStore:
scope_tree = discover_scopes(type(config))
self.registry.validate_scopes({scope.name for scope in scope_tree})
state = _WalkState(
config=config,
store=BuildStore(scope_tree=scope_tree),
ops=self.registry.sorted_by_scope(),
scope_tree=scope_tree,
package_prefix=self.package_prefix,
)
_visit(
scope=PROJECT,
instance_config=config,
instance_id=NodePath(("project",)),
state=state,
)
return state.store
@dataclass
class _WalkState:
config: BaseModel
store: BuildStore
ops: dict[str, list[OperationEntry]]
scope_tree: ScopeTree
package_prefix: str
def _visit(
scope: Scope,
instance_config: BaseModel,
instance_id: NodePath,
state: _WalkState,
) -> None:
state.store.register_instance(
instance_id,
instance_config,
)
ctx = BuildContext(
config=state.config,
scope=scope,
instance=instance_config,
instance_id=instance_id,
store=state.store,
package_prefix=state.package_prefix,
)
ops = state.ops.get(scope.name, [])
_run_ops(ops, ctx, after_children=False)
for child_scope in state.scope_tree.children_of(scope):
scope_configs = _resolve_scope_list(
scope=child_scope,
parent_config=instance_config,
)
index = 0
while index < len(scope_configs):
_visit(
scope=child_scope,
instance_config=scope_configs[index],
instance_id=NodePath.combine(
instance_id, f"{child_scope.config_key}.{index}"
),
state=state,
)
index += 1
_run_ops(ops, ctx, after_children=True)
def _run_ops(
ops: list[OperationEntry],
ctx: BuildContext[Any, BaseModel],
*,
after_children: bool,
) -> None:
for meta, op_cls in ops:
if meta.after_children != after_children:
continue
# dispatch_on: fire only on the instance whose discriminator
# matches this op's name (e.g. OperationConfig.name == "get").
if meta.dispatch_on is not None and getattr(
ctx.instance, meta.dispatch_on, None
) != (meta.match_value or meta.name):
continue
operation_instance = op_cls()
when_method = getattr(operation_instance, "when", None)
if callable(when_method) and not when_method(ctx):
continue
options = _resolve_options(op_cls, ctx.instance)
ctx.store.add(
ctx.instance_id,
meta.name,
*operation_instance.build(ctx, options),
)
def _resolve_scope_list(
*,
scope: Scope,
parent_config: BaseModel,
) -> list[BaseModel]:
attr_value: object = parent_config
for attr in scope.resolve_path:
attr_value = getattr(attr_value, attr)
return cast("list[BaseModel]", attr_value)
def _resolve_options(op_cls: type, instance: object) -> BaseModel:
options_cls = getattr(op_cls, "Options", None)
if options_cls is None:
return EmptyOptions()
if isinstance(instance, BaseModel):
raw = getattr(instance, "options", None)
if isinstance(raw, dict):
return options_cls(**raw)
return options_cls()