from typing import Any, Dict, Generator, Tuple
from taskgraph.util.attributes import keymatch
def iter_dot_path(
container: Dict[str, Any], subfield: str
) -> Generator[Tuple[Dict[str, Any], str], None, None]:
while "." in subfield:
f, subfield = subfield.split(".", 1)
if f.endswith("[]"):
f = f[0:-2]
if not isinstance(container.get(f), list):
return
for item in container[f]:
yield from iter_dot_path(item, subfield)
return
if f not in container:
return
container = container[f]
if isinstance(container, dict) and subfield in container:
yield container, subfield
def evaluate_keyed_by(
value, item_name, attributes, defer=None, enforce_single_match=True
):
while True:
if not isinstance(value, dict) or len(value) != 1:
return value
value_key = next(iter(value))
if not value_key.startswith("by-"):
return value
keyed_by = value_key[3:]
if defer and keyed_by in defer:
return value
key = attributes.get(keyed_by)
alternatives = next(iter(value.values()))
if len(alternatives) == 1 and "default" in alternatives:
raise Exception(
f"Keyed-by '{keyed_by}' unnecessary with only value 'default' "
f"found, when determining item {item_name}"
)
if key is None:
if "default" in alternatives:
value = alternatives["default"]
continue
else:
raise Exception(
f"No attribute {keyed_by} and no value for 'default' found "
f"while determining item {item_name}"
)
matches = keymatch(alternatives, key)
if enforce_single_match and len(matches) > 1:
raise Exception(
f"Multiple matching values for {keyed_by} {key!r} found while "
f"determining item {item_name}"
)
elif matches:
value = matches[0]
continue
raise Exception(
f"No {keyed_by} matching {key!r} nor 'default' found while determining item {item_name}"
)