| 19 | |
| 20 | |
| 21 | class Environment: |
| 22 | def __init__(self, config: "Config", prefix: str) -> None: |
| 23 | self._config = config |
| 24 | self._prefix = prefix |
| 25 | self.data: Dict[str, Any] = {} # Accumulator |
| 26 | |
| 27 | def load(self) -> Dict[str, Any]: |
| 28 | """ |
| 29 | Return a nested dict containing values from `os.environ`. |
| 30 | |
| 31 | Specifically, values whose keys map to already-known configuration |
| 32 | settings, allowing us to perform basic typecasting. |
| 33 | |
| 34 | See :ref:`env-vars` for details. |
| 35 | """ |
| 36 | # Obtain allowed env var -> existing value map |
| 37 | env_vars = self._crawl(key_path=[], env_vars={}) |
| 38 | m = "Scanning for env vars according to prefix: {!r}, mapping: {!r}" |
| 39 | debug(m.format(self._prefix, env_vars)) |
| 40 | # Check for actual env var (honoring prefix) and try to set |
| 41 | for env_var, key_path in env_vars.items(): |
| 42 | real_var = (self._prefix or "") + env_var |
| 43 | if real_var in os.environ: |
| 44 | self._path_set(key_path, os.environ[real_var]) |
| 45 | debug("Obtained env var config: {!r}".format(self.data)) |
| 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: |
no outgoing calls
no test coverage detected
searching dependent graphs…