1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
|
import os
import string
import warnings
from typing import Any, Optional
from omegaconf import Container, Node
from omegaconf._utils import _DEFAULT_MARKER_, _get_value
from omegaconf.basecontainer import BaseContainer
from omegaconf.errors import ConfigKeyError
from omegaconf.grammar_parser import parse
from omegaconf.resolvers.oc import dict
def create(obj: Any, _parent_: Container) -> Any:
"""Create a config object from `obj`, similar to `OmegaConf.create`"""
from omegaconf import OmegaConf
assert isinstance(_parent_, BaseContainer)
return OmegaConf.create(obj, parent=_parent_)
def env(key: str, default: Any = _DEFAULT_MARKER_) -> Optional[str]:
"""
:param key: Environment variable key
:param default: Optional default value to use in case the key environment variable is not set.
If default is not a string, it is converted with str(default).
None default is returned as is.
:return: The environment variable 'key'. If the environment variable is not set and a default is
provided, the default is used. If used, the default is converted to a string with str(default).
If the default is None, None is returned (without a string conversion).
"""
try:
return os.environ[key]
except KeyError:
if default is not _DEFAULT_MARKER_:
return str(default) if default is not None else None
else:
raise KeyError(f"Environment variable '{key}' not found")
def decode(expr: Optional[str], _parent_: Container, _node_: Node) -> Any:
"""
Parse and evaluate `expr` according to the `singleElement` rule of the grammar.
If `expr` is `None`, then return `None`.
"""
if expr is None:
return None
if not isinstance(expr, str):
raise TypeError(
f"`oc.decode` can only take strings or None as input, "
f"but `{expr}` is of type {type(expr).__name__}"
)
parse_tree = parse(expr, parser_rule="singleElement", lexer_mode="VALUE_MODE")
val = _parent_.resolve_parse_tree(parse_tree, node=_node_)
return _get_value(val)
def deprecated(
key: str,
message: str = "'$OLD_KEY' is deprecated. Change your code and config to use '$NEW_KEY'",
*,
_parent_: Container,
_node_: Node,
) -> Any:
from omegaconf._impl import select_node
if not isinstance(key, str):
raise TypeError(
f"oc.deprecated: interpolation key type is not a string ({type(key).__name__})"
)
if not isinstance(message, str):
raise TypeError(
f"oc.deprecated: interpolation message type is not a string ({type(message).__name__})"
)
full_key = _node_._get_full_key(key=None)
target_node = select_node(_parent_, key, absolute_key=True)
if target_node is None:
raise ConfigKeyError(
f"In oc.deprecated resolver at '{full_key}': Key not found: '{key}'"
)
new_key = target_node._get_full_key(key=None)
msg = string.Template(message).safe_substitute(
OLD_KEY=full_key,
NEW_KEY=new_key,
)
warnings.warn(category=UserWarning, message=msg)
return target_node
def select(
key: str,
default: Any = _DEFAULT_MARKER_,
*,
_parent_: Container,
) -> Any:
from omegaconf._impl import select_value
return select_value(cfg=_parent_, key=key, absolute_key=True, default=default)
__all__ = [
"create",
"decode",
"deprecated",
"dict",
"env",
"select",
]
|
