blueprints: Resolve yamltags in state and model attributes (#4299)

* Fixed state and model attributes not resolving yaml tags

* Linting
This commit is contained in:
sdimovv 2022-12-29 09:05:32 +00:00 committed by GitHub
parent a6755bea71
commit ce5d1fd80d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 21 additions and 9 deletions

View file

@ -4,7 +4,8 @@ context:
policy_property: name policy_property: name
policy_property_value: foo-bar-baz-qux policy_property_value: foo-bar-baz-qux
entries: entries:
- model: authentik_sources_oauth.oauthsource - model: !Format ["%s", authentik_sources_oauth.oauthsource]
state: !Format ["%s", present]
identifiers: identifiers:
slug: test slug: test
attrs: attrs:

View file

@ -5,7 +5,7 @@ from enum import Enum
from functools import reduce from functools import reduce
from operator import ixor from operator import ixor
from os import getenv from os import getenv
from typing import Any, Literal, Optional from typing import Any, Literal, Optional, Union
from uuid import UUID from uuid import UUID
from django.apps import apps from django.apps import apps
@ -56,8 +56,10 @@ class BlueprintEntryDesiredState(Enum):
class BlueprintEntry: class BlueprintEntry:
"""Single entry of a blueprint""" """Single entry of a blueprint"""
model: str model: Union[str, "YAMLTag"]
state: BlueprintEntryDesiredState = field(default=BlueprintEntryDesiredState.PRESENT) state: Union[BlueprintEntryDesiredState, "YAMLTag"] = field(
default=BlueprintEntryDesiredState.PRESENT
)
conditions: list[Any] = field(default_factory=list) conditions: list[Any] = field(default_factory=list)
identifiers: dict[str, Any] = field(default_factory=dict) identifiers: dict[str, Any] = field(default_factory=dict)
attrs: Optional[dict[str, Any]] = field(default_factory=dict) attrs: Optional[dict[str, Any]] = field(default_factory=dict)
@ -103,6 +105,14 @@ class BlueprintEntry:
"""Get attributes of this entry, with all yaml tags resolved""" """Get attributes of this entry, with all yaml tags resolved"""
return self.tag_resolver(self.identifiers, blueprint) return self.tag_resolver(self.identifiers, blueprint)
def get_state(self, blueprint: "Blueprint") -> BlueprintEntryDesiredState:
"""Get the blueprint state, with yaml tags resolved if present"""
return BlueprintEntryDesiredState(self.tag_resolver(self.state, blueprint))
def get_model(self, blueprint: "Blueprint") -> str:
"""Get the blueprint model, with yaml tags resolved if present"""
return str(self.tag_resolver(self.model, blueprint))
def check_all_conditions_match(self, blueprint: "Blueprint") -> bool: def check_all_conditions_match(self, blueprint: "Blueprint") -> bool:
"""Check all conditions of this entry match (evaluate to True)""" """Check all conditions of this entry match (evaluate to True)"""
return all(self.tag_resolver(self.conditions, blueprint)) return all(self.tag_resolver(self.conditions, blueprint))

View file

@ -148,7 +148,7 @@ class Importer:
self.logger.debug("One or more conditions of this entry are not fulfilled, skipping") self.logger.debug("One or more conditions of this entry are not fulfilled, skipping")
return None return None
model_app_label, model_name = entry.model.split(".") model_app_label, model_name = entry.get_model(self.__import).split(".")
model: type[SerializerModel] = registry.get_model(model_app_label, model_name) model: type[SerializerModel] = registry.get_model(model_app_label, model_name)
# Don't use isinstance since we don't want to check for inheritance # Don't use isinstance since we don't want to check for inheritance
if not is_model_allowed(model): if not is_model_allowed(model):
@ -184,7 +184,7 @@ class Importer:
serializer_kwargs = {} serializer_kwargs = {}
model_instance = existing_models.first() model_instance = existing_models.first()
if not isinstance(model(), BaseMetaModel) and model_instance: if not isinstance(model(), BaseMetaModel) and model_instance:
if entry.state == BlueprintEntryDesiredState.CREATED: if entry.get_state(self.__import) == BlueprintEntryDesiredState.CREATED:
self.logger.debug("instance exists, skipping") self.logger.debug("instance exists, skipping")
return None return None
self.logger.debug( self.logger.debug(
@ -237,7 +237,7 @@ class Importer:
"""Apply (create/update) models yaml""" """Apply (create/update) models yaml"""
self.__pk_map = {} self.__pk_map = {}
for entry in self.__import.entries: for entry in self.__import.entries:
model_app_label, model_name = entry.model.split(".") model_app_label, model_name = entry.get_model(self.__import).split(".")
try: try:
model: type[SerializerModel] = registry.get_model(model_app_label, model_name) model: type[SerializerModel] = registry.get_model(model_app_label, model_name)
except LookupError: except LookupError:
@ -254,7 +254,8 @@ class Importer:
if not serializer: if not serializer:
continue continue
if entry.state in [ state = entry.get_state(self.__import)
if state in [
BlueprintEntryDesiredState.PRESENT, BlueprintEntryDesiredState.PRESENT,
BlueprintEntryDesiredState.CREATED, BlueprintEntryDesiredState.CREATED,
]: ]:
@ -263,7 +264,7 @@ class Importer:
self.__pk_map[entry.identifiers["pk"]] = model.pk self.__pk_map[entry.identifiers["pk"]] = model.pk
entry._state = BlueprintEntryState(model) entry._state = BlueprintEntryState(model)
self.logger.debug("updated model", model=model) self.logger.debug("updated model", model=model)
elif entry.state == BlueprintEntryDesiredState.ABSENT: elif state == BlueprintEntryDesiredState.ABSENT:
instance: Optional[Model] = serializer.instance instance: Optional[Model] = serializer.instance
if instance: if instance:
instance.delete() instance.delete()