MCPcopy Index your code
hub / github.com/pyinvoke/invoke / _crawl

Method _crawl

invoke/env.py:48–86  ·  view source on GitHub ↗

Examine config at location ``key_path`` & return potential env vars. Uses ``env_vars`` dict to determine if a conflict exists, and raises an exception if so. This dict is of the following form:: { 'EXPECTED_ENV_VAR_HERE': ['actual', 'nested', 'k

(
        self, key_path: List[str], env_vars: Mapping[str, Sequence[str]]
    )

Source from the content-addressed store, hash-verified

46 return self.data
47
48 def _crawl(
49 self, key_path: List[str], env_vars: Mapping[str, Sequence[str]]
50 ) -> Dict[str, Any]:
51 """
52 Examine config at location ``key_path`` & return potential env vars.
53
54 Uses ``env_vars`` dict to determine if a conflict exists, and raises an
55 exception if so. This dict is of the following form::
56
57 {
58 'EXPECTED_ENV_VAR_HERE': ['actual', 'nested', 'key_path'],
59 ...
60 }
61
62 Returns another dictionary of new keypairs as per above.
63 """
64 new_vars: Dict[str, List[str]] = {}
65 obj = self._path_get(key_path)
66 # Sub-dict -> recurse
67 if (
68 hasattr(obj, "keys")
69 and callable(obj.keys)
70 and hasattr(obj, "__getitem__")
71 ):
72 for key in obj.keys():
73 merged_vars = dict(env_vars, **new_vars)
74 merged_path = key_path + [key]
75 crawled = self._crawl(merged_path, merged_vars)
76 # Handle conflicts
77 for key in crawled:
78 if key in new_vars:
79 err = "Found >1 source for {}"
80 raise AmbiguousEnvVar(err.format(key))
81 # Merge and continue
82 new_vars.update(crawled)
83 # Other -> is leaf, no recursion
84 else:
85 new_vars[self._to_env_var(key_path)] = key_path
86 return new_vars
87
88 def _to_env_var(self, key_path: Iterable[str]) -> str:
89 return "_".join(key_path).upper()

Callers 1

loadMethod · 0.95

Calls 4

_path_getMethod · 0.95
_to_env_varMethod · 0.95
AmbiguousEnvVarClass · 0.85
updateMethod · 0.45

Tested by

no test coverage detected