99 lines
2.5 KiB
Python
99 lines
2.5 KiB
Python
from collections.abc import Iterable
|
|
|
|
from .adapters import PlanAction
|
|
|
|
|
|
def _action_identity(
|
|
action: PlanAction,
|
|
) -> tuple[str, str, str]:
|
|
desired = action.desired
|
|
if desired is None:
|
|
raise RuntimeError(
|
|
"Dependency-ordered plan action requires desired state"
|
|
)
|
|
|
|
return (
|
|
action.endpoint_id,
|
|
desired.object_type,
|
|
desired.logical_key,
|
|
)
|
|
|
|
|
|
def order_plan_actions(
|
|
actions: Iterable[PlanAction],
|
|
) -> tuple[PlanAction, ...]:
|
|
action_list = list(actions)
|
|
by_identity: dict[
|
|
tuple[str, str, str],
|
|
PlanAction,
|
|
] = {}
|
|
positions: dict[
|
|
tuple[str, str, str],
|
|
int,
|
|
] = {}
|
|
|
|
for position, action in enumerate(action_list):
|
|
identity = _action_identity(action)
|
|
if identity in by_identity:
|
|
raise RuntimeError(
|
|
"Duplicate plan action identity: "
|
|
f"{identity[0]} {identity[1]} {identity[2]!r}"
|
|
)
|
|
by_identity[identity] = action
|
|
positions[identity] = position
|
|
|
|
dependencies: dict[
|
|
tuple[str, str, str],
|
|
set[tuple[str, str, str]],
|
|
] = {
|
|
identity: set()
|
|
for identity in by_identity
|
|
}
|
|
|
|
for identity, action in by_identity.items():
|
|
desired = action.desired
|
|
for reference in desired.references:
|
|
if reference.logical_key is None:
|
|
continue
|
|
|
|
dependency = (
|
|
action.endpoint_id,
|
|
reference.object_type,
|
|
reference.logical_key,
|
|
)
|
|
if dependency in by_identity:
|
|
dependencies[identity].add(dependency)
|
|
|
|
remaining = set(by_identity)
|
|
ordered: list[PlanAction] = []
|
|
|
|
while remaining:
|
|
ready = [
|
|
identity
|
|
for identity in remaining
|
|
if not dependencies[identity].intersection(remaining)
|
|
]
|
|
|
|
if not ready:
|
|
cycle = ", ".join(
|
|
f"{identity[1]} {identity[2]!r}"
|
|
for identity in sorted(
|
|
remaining,
|
|
key=lambda item: positions[item],
|
|
)
|
|
)
|
|
raise RuntimeError(
|
|
"Plan action dependency cycle: "
|
|
f"{cycle}"
|
|
)
|
|
|
|
ready.sort(
|
|
key=lambda identity: positions[identity]
|
|
)
|
|
|
|
for identity in ready:
|
|
ordered.append(by_identity[identity])
|
|
remaining.remove(identity)
|
|
|
|
return tuple(ordered)
|