from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from dxaws_core.o11y import O11y
from dxaws_website.executor import ExecutionRuntime, apply_plan
from dxaws_website.models import WebsiteCurrent, WebsiteDesired, WebsiteOutputs, WebsitePlan, WebsiteResult
from dxaws_website.planner import build_plan
[docs]
@dataclass(frozen=True, slots=True, kw_only=True)
class ApplyOptions:
"""Options controlling a single executor apply run."""
emit_events: bool = True
[docs]
@dataclass(frozen=True, slots=True, kw_only=True)
class ExecuteOptions:
"""Options for execute() (plan + apply)."""
apply_options: ApplyOptions | None = None
emit_events: bool = True
[docs]
@dataclass(frozen=True, slots=True, kw_only=True)
class WebsiteManager:
"""Thin orchestration surface for dxaws-website.
The manager owns the public website API. Planner decides what to do,
executor follows the plan, and adapters are the only boundary to primitive
modules.
"""
runtime: ExecutionRuntime
o11y: O11y = field(default_factory=O11y.noop)
def _emit(self, event: str, **fields: Any) -> None:
self.o11y.info(event, **fields)
[docs]
def get_component_statuses(self, desired: WebsiteDesired) -> dict[str, Any]:
"""Collect normalized website-layer status from the execution runtime.
Manager should not know which adapters exist or how their status is
gathered. That is an execution/runtime concern.
"""
getter = getattr(self.runtime, "get_component_statuses", None)
if not callable(getter):
raise RuntimeError("ExecutionRuntime must provide get_component_statuses(desired)")
return getter(desired)
[docs]
def get_current(self, desired: WebsiteDesired) -> WebsiteCurrent:
"""Get current website state from the execution runtime.
Manager should not assemble component state itself. Runtime/executor-side
code owns the adapter interactions needed to produce website current
state.
"""
getter = getattr(self.runtime, "get_current", None)
if not callable(getter):
raise RuntimeError("ExecutionRuntime must provide get_current(desired)")
return getter(desired)
[docs]
def plan(self, desired: WebsiteDesired, current: WebsiteCurrent | None = None) -> WebsitePlan:
current = current or self.get_current(desired)
component_statuses = self.get_component_statuses(desired)
self._emit(
"manager.plan.start",
module="dxaws-website",
url=desired.url,
)
plan = build_plan(
desired=desired,
current=current,
component_statuses=component_statuses,
)
self._emit(
"manager.plan.done",
module="dxaws-website",
url=desired.url,
action=plan.action,
steps=plan.steps,
)
return plan
[docs]
def apply(self, plan: WebsitePlan, *, options: ApplyOptions | None = None) -> WebsiteOutputs:
options = options or ApplyOptions()
return apply_plan(
self.runtime,
plan,
options=options,
emit=self._emit,
)
[docs]
def execute(self, desired: WebsiteDesired, *, options: ExecuteOptions | None = None) -> WebsiteResult:
options = options or ExecuteOptions()
apply_options = options.apply_options or ApplyOptions()
if options.emit_events:
self._emit(
"manager.execute.start",
module="dxaws-website",
url=desired.url,
)
current = self.get_current(desired)
plan = self.plan(desired, current)
if plan.action in {"create", "update"}:
outputs = self.apply(plan, options=apply_options)
outcome = "applied"
elif plan.action == "noop":
outputs = WebsiteOutputs()
outcome = "noop"
elif plan.action == "wait":
outputs = WebsiteOutputs()
outcome = "wait"
else:
raise RuntimeError(f"Unknown plan action: {plan.action}")
refreshed_current = self.get_current(desired)
if options.emit_events:
self._emit(
"manager.execute.done",
module="dxaws-website",
url=desired.url,
action=plan.action,
outcome=outcome,
)
return WebsiteResult(
desired=desired,
current=refreshed_current,
plan=plan,
outputs=outputs,
outcome=outcome,
)