Skip to content

Commit

Permalink
Resolve nested secrets without validation
Browse files Browse the repository at this point in the history
  • Loading branch information
jvansanten committed Feb 6, 2025
1 parent 5122548 commit 8550583
Showing 1 changed file with 18 additions and 12 deletions.
30 changes: 18 additions & 12 deletions ampel/core/UnitLoader.py
Original file line number Diff line number Diff line change
Expand Up @@ -347,7 +347,7 @@ def resolve_aliases(self, value):
def resolve_secrets(self, unit_type: type[AmpelUnit], init_kwargs: dict[str, Any]) -> dict[str, Any]:
"""
Add a resolved Secret instance to init_kwargs for every Secret field of
unit_type.
unit_type, recursing into nested UnitModels.
"""
for k, annotation in unit_type._annots.items(): # noqa: SLF001
# for unions, consider the first member that is not NoneType
Expand All @@ -359,7 +359,7 @@ def resolve_secrets(self, unit_type: type[AmpelUnit], init_kwargs: dict[str, Any
continue
annotation = next((f for f in get_args(annotation) if f is not type(None)), type(None)) # noqa: PLW2901
field_type = get_origin(annotation) or annotation
if issubclass(type(field_type), type) and issubclass(field_type, Secret):
if issubclass(type(field_type), type) and issubclass(field_type, Secret|UnitModel):
default = False
if isinstance(kwargs := init_kwargs.get(k), Mapping):
v = field_type(**kwargs)
Expand All @@ -369,16 +369,22 @@ def resolve_secrets(self, unit_type: type[AmpelUnit], init_kwargs: dict[str, Any
else:
# missing required field; will be caught in validation later
continue
ValueType = args[0] if (args := annotation.get_model_args()) else object
if args:
assert ValueType is not object
if not self.vault:
raise TypeError("No vault configured")
if not self.vault.resolve_secret(v, ValueType):
raise TypeError(
f"Could not resolve {unit_type.__name__}.{k} as {getattr(ValueType, '__name__', '<untyped>')}"
f" using {'default' if default else 'configured'} value {v!r}"
)
if issubclass(field_type, Secret):
ValueType = args[0] if (args := annotation.get_model_args()) else object
if args:
assert ValueType is not object
if not self.vault:
raise TypeError("No vault configured")
if not self.vault.resolve_secret(v, ValueType):
raise TypeError(
f"Could not resolve {unit_type.__name__}.{k} as {getattr(ValueType, '__name__', '<untyped>')}"
f" using {'default' if default else 'configured'} value {v!r}"
)
elif isinstance(v, UnitModel) and isinstance(v.config, Mapping | None):
v.config = self.resolve_secrets(
self.get_class_by_name(v.unit),
(v.config or {}) | (v.override or {})
)
init_kwargs[k] = v

return init_kwargs
Expand Down

0 comments on commit 8550583

Please sign in to comment.