blueprints: Added conditional entry application (#4167)
* blueprints: Added !AsBool tag * Renamed AsBool tag to Condition * Added conditions attributed to BlueprintEntry * Added docs for the conditions attribute of a blueprint entry * Website linting fix * add new tag to vscode settings Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org> Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org> Co-authored-by: Jens Langhammer <jens.langhammer@beryju.org>
This commit is contained in:
@ -2,7 +2,9 @@
|
||||
from collections import OrderedDict
|
||||
from dataclasses import asdict, dataclass, field, is_dataclass
|
||||
from enum import Enum
|
||||
from typing import Any, Optional
|
||||
from functools import reduce
|
||||
from operator import ixor
|
||||
from typing import Any, Literal, Optional
|
||||
from uuid import UUID
|
||||
|
||||
from django.apps import apps
|
||||
@ -55,6 +57,7 @@ class BlueprintEntry:
|
||||
|
||||
model: str
|
||||
state: BlueprintEntryDesiredState = field(default=BlueprintEntryDesiredState.PRESENT)
|
||||
conditions: list[Any] = field(default_factory=list)
|
||||
identifiers: dict[str, Any] = field(default_factory=dict)
|
||||
attrs: Optional[dict[str, Any]] = field(default_factory=dict)
|
||||
|
||||
@ -99,6 +102,10 @@ class BlueprintEntry:
|
||||
"""Get attributes of this entry, with all yaml tags resolved"""
|
||||
return self.tag_resolver(self.identifiers, blueprint)
|
||||
|
||||
def check_all_conditions_match(self, blueprint: "Blueprint") -> bool:
|
||||
"""Check all conditions of this entry match (evaluate to True)"""
|
||||
return all(self.tag_resolver(self.conditions, blueprint))
|
||||
|
||||
|
||||
@dataclass
|
||||
class BlueprintMetadata:
|
||||
@ -241,6 +248,49 @@ class Find(YAMLTag):
|
||||
return None
|
||||
|
||||
|
||||
class Condition(YAMLTag):
|
||||
"""Convert all values to a single boolean"""
|
||||
|
||||
mode: Literal["AND", "NAND", "OR", "NOR", "XOR", "XNOR"]
|
||||
args: list[Any]
|
||||
|
||||
_COMPARATORS = {
|
||||
# Using all and any here instead of from operator import iand, ior
|
||||
# to improve performance
|
||||
"AND": all,
|
||||
"NAND": lambda args: not all(args),
|
||||
"OR": any,
|
||||
"NOR": lambda args: not any(args),
|
||||
"XOR": lambda args: reduce(ixor, args) if len(args) > 1 else args[0],
|
||||
"XNOR": lambda args: not (reduce(ixor, args) if len(args) > 1 else args[0]),
|
||||
}
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
def __init__(self, loader: "BlueprintLoader", node: SequenceNode) -> None:
|
||||
super().__init__()
|
||||
self.mode = node.value[0].value
|
||||
self.args = []
|
||||
for raw_node in node.value[1:]:
|
||||
self.args.append(loader.construct_object(raw_node))
|
||||
|
||||
def resolve(self, entry: BlueprintEntry, blueprint: Blueprint) -> Any:
|
||||
args = []
|
||||
for arg in self.args:
|
||||
if isinstance(arg, YAMLTag):
|
||||
args.append(arg.resolve(entry, blueprint))
|
||||
else:
|
||||
args.append(arg)
|
||||
|
||||
if not args:
|
||||
raise EntryInvalidError("At least one value is required after mode selection.")
|
||||
|
||||
try:
|
||||
comparator = self._COMPARATORS[self.mode.upper()]
|
||||
return comparator(tuple(bool(x) for x in args))
|
||||
except (TypeError, KeyError) as exc:
|
||||
raise EntryInvalidError(exc)
|
||||
|
||||
|
||||
class BlueprintDumper(SafeDumper):
|
||||
"""Dump dataclasses to yaml"""
|
||||
|
||||
@ -281,6 +331,7 @@ class BlueprintLoader(SafeLoader):
|
||||
self.add_constructor("!Find", Find)
|
||||
self.add_constructor("!Context", Context)
|
||||
self.add_constructor("!Format", Format)
|
||||
self.add_constructor("!Condition", Condition)
|
||||
|
||||
|
||||
class EntryInvalidError(SentryIgnoredException):
|
||||
|
Reference in New Issue
Block a user