__init__.py 3.5 KB

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