"""OmegaConf module"""
import copy
import inspect
import io
import os
import pathlib
import re
import sys
import warnings
from collections import defaultdict
from contextlib import contextmanager, nullcontext
from enum import Enum
from textwrap import dedent
from typing import (
IO,
Any,
Callable,
Dict,
Generator,
Iterable,
List,
Optional,
Set,
Tuple,
Type,
Union,
overload,
)
import yaml
from . import DictConfig, DictKeyType, ListConfig
from ._utils import (
_DEFAULT_MARKER_,
_ensure_container,
_get_value,
format_and_raise,
get_dict_key_value_types,
get_list_element_type,
get_omega_conf_dumper,
get_type_of,
is_attr_class,
is_dataclass,
is_dict_annotation,
is_int,
is_list_annotation,
is_literal_annotation,
is_primitive_container,
is_primitive_dict,
is_primitive_list,
is_structured_config,
is_tuple_annotation,
is_union_annotation,
split_key,
type_str,
)
from ._yaml import _DEFAULT_MAX_YAML_EXPANDED_NODES, get_yaml_loader
from .base import Box, Container, ListMergeMode, Node, SCMode, UnionNode
from .basecontainer import BaseContainer
from .errors import (
ConfigTypeError,
InterpolationResolutionError,
InterpolationToMissingValueError,
MissingMandatoryValue,
OmegaConfBaseException,
UnsupportedInterpolationType,
ValidationError,
)
from .nodes import (
AnyNode,
BooleanNode,
BytesNode,
EnumNode,
FloatNode,
IntegerNode,
LiteralNode,
PathNode,
StringNode,
ValueNode,
)
MISSING: Any = "???"
Resolver = Callable[..., Any]
_CUSTOM_RESOLVER_INTERPOLATION_PATTERN = re.compile(r"\${[^}]*:")
def II(interpolation: str) -> Any:
"""
Equivalent to ``${interpolation}``
:param interpolation:
:return: input ``${node}`` with type Any
"""
return "${" + interpolation + "}"
def SI(interpolation: str) -> Any:
"""
Use this for String interpolation, for example ``"http://${host}:${port}"``
:param interpolation: interpolation string
:return: input interpolation with type ``Any``
"""
return interpolation
def register_default_resolvers() -> None:
from omegaconf.resolvers import oc
OmegaConf.register_resolver("oc.create", oc.create)
OmegaConf.register_resolver("oc.decode", oc.decode)
OmegaConf.register_resolver("oc.deprecated", oc.deprecated)
OmegaConf.register_resolver("oc.env", oc.env)
OmegaConf.register_resolver("oc.select", oc.select)
OmegaConf.register_resolver("oc.dict.keys", oc.dict.keys)
OmegaConf.register_resolver("oc.dict.values", oc.dict.values)
[docs]
class OmegaConf:
"""OmegaConf primary class"""
def __init__(self) -> None:
raise NotImplementedError("Use one of the static construction functions")
[docs]
@staticmethod
def structured(
obj: Any,
parent: Optional[BaseContainer] = None,
flags: Optional[Dict[str, bool]] = None,
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> Any:
"""
Alias for ``OmegaConf.create(obj)``. Accepts any input that ``create`` accepts,
though intended for structured config objects (dataclass or attrs types/instances).
:param obj: Source object — typically a dataclass or attrs type or instance,
but any value accepted by ``OmegaConf.create`` is valid.
:param parent: Optional parent node.
:param flags: Optional flags dict (e.g. ``{"readonly": True}``).
:param max_yaml_expanded_nodes: Maximum YAML nodes after alias expansion
when ``obj`` is a YAML string. By default, OmegaConf uses the
``OMEGACONF_MAX_YAML_EXPANDED_NODES`` environment variable if set,
otherwise ``10_000``. Explicit arguments override the environment.
Pass ``None`` only for trusted input. See
https://omegaconf.readthedocs.io/en/latest/yaml_aliases.html.
:return: A ``DictConfig``, ``ListConfig``, or ``None`` (when ``obj`` is ``None``).
"""
return OmegaConf.create(
obj, parent, flags, max_yaml_expanded_nodes=max_yaml_expanded_nodes
)
@staticmethod
@overload
def create(
obj: str,
parent: Optional[BaseContainer] = None,
flags: Optional[Dict[str, bool]] = None,
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> Union[DictConfig, ListConfig]: ...
@staticmethod
@overload
def create(
obj: Union[List[Any], Tuple[Any, ...]],
parent: Optional[BaseContainer] = None,
flags: Optional[Dict[str, bool]] = None,
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> ListConfig: ...
@staticmethod
@overload
def create(
obj: DictConfig,
parent: Optional[BaseContainer] = None,
flags: Optional[Dict[str, bool]] = None,
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> DictConfig: ...
@staticmethod
@overload
def create(
obj: ListConfig,
parent: Optional[BaseContainer] = None,
flags: Optional[Dict[str, bool]] = None,
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> ListConfig: ...
@staticmethod
@overload
def create(
obj: None,
parent: Optional[BaseContainer] = None,
flags: Optional[Dict[str, bool]] = None,
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> None: ...
@staticmethod
@overload
def create(
obj: Dict[Any, Any] = ...,
parent: Optional[BaseContainer] = None,
flags: Optional[Dict[str, bool]] = None,
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> DictConfig: ...
[docs]
@staticmethod
def create( # noqa F811
obj: Any = _DEFAULT_MARKER_,
parent: Optional[BaseContainer] = None,
flags: Optional[Dict[str, bool]] = None,
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> Optional[Union[DictConfig, ListConfig]]:
"""
Create an OmegaConf config from ``obj``.
``obj`` may be a YAML string, a dict, a list or tuple, a dataclass or attrs class (type or
instance), an existing ``DictConfig`` / ``ListConfig``, or ``None``.
Omitting ``obj`` (or passing ``{}`` explicitly) returns an empty ``DictConfig``.
:param obj: Source object to build the config from.
:param parent: Optional parent node.
:param flags: Optional flags dict (e.g. ``{"readonly": True}``).
:param max_yaml_expanded_nodes: Maximum YAML nodes after alias expansion
when ``obj`` is a YAML string. By default, OmegaConf uses the
``OMEGACONF_MAX_YAML_EXPANDED_NODES`` environment variable if set,
otherwise ``10_000``. Explicit arguments override the environment.
Pass ``None`` only for trusted input. See
https://omegaconf.readthedocs.io/en/latest/yaml_aliases.html.
:return: A ``DictConfig``, ``ListConfig``, or ``None`` (when ``obj`` is ``None``).
"""
return OmegaConf._create_impl(
obj=obj,
parent=parent,
flags=flags,
max_yaml_expanded_nodes=max_yaml_expanded_nodes,
)
[docs]
@staticmethod
def typed_list(
content: Optional[List[Any]] = None,
element_type: Any = Any,
) -> ListConfig:
"""Create a ListConfig with an explicit element type.
Useful for disambiguating assignment to a Union[List[X], List[Y]] field
when the value is empty or otherwise matches multiple candidates.
"""
from typing import List as _List
ref_type = _List[element_type]
return ListConfig(
content=content if content is not None else [],
element_type=element_type,
ref_type=ref_type,
)
[docs]
@staticmethod
def typed_dict(
content: Optional[Dict[Any, Any]] = None,
key_type: Any = Any,
element_type: Any = Any,
) -> DictConfig:
"""Create a DictConfig with explicit key and value types.
Useful for disambiguating assignment to a Union[Dict[str, X], Dict[str, Y]]
field when the value is empty or otherwise matches multiple candidates.
"""
from typing import Dict as _Dict
ref_type = _Dict[key_type, element_type]
return DictConfig(
content=content if content is not None else {},
key_type=key_type,
element_type=element_type,
ref_type=ref_type,
)
[docs]
@staticmethod
def load(
file_: Union[str, pathlib.Path, IO[Any]],
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> Union[DictConfig, ListConfig]:
"""
Load a YAML config from a file path or file-like object.
:param file_: A file path (str or ``pathlib.Path``) or an open file object.
:param max_yaml_expanded_nodes: Maximum YAML nodes after alias expansion.
By default, OmegaConf uses the
``OMEGACONF_MAX_YAML_EXPANDED_NODES`` environment variable if set,
otherwise ``10_000``. Explicit arguments override the environment.
Pass ``None`` only for trusted input. See
https://omegaconf.readthedocs.io/en/latest/yaml_aliases.html.
:return: A ``DictConfig`` or ``ListConfig`` parsed from the YAML content.
"""
if isinstance(file_, (str, pathlib.Path)):
with io.open(os.path.abspath(file_), "r", encoding="utf-8") as f:
obj = yaml.load(
f,
Loader=get_yaml_loader(
max_yaml_expanded_nodes=max_yaml_expanded_nodes
),
)
elif getattr(file_, "read", None):
obj = yaml.load(
file_,
Loader=get_yaml_loader(max_yaml_expanded_nodes=max_yaml_expanded_nodes),
)
else:
raise TypeError("Unexpected file type")
if obj is not None and not isinstance(obj, (list, dict, str)):
raise IOError( # pragma: no cover
f"Invalid loaded object type: {type(obj).__name__}"
)
ret: Union[DictConfig, ListConfig]
if obj is None:
ret = OmegaConf.create()
else:
ret = OmegaConf.create(obj, max_yaml_expanded_nodes=max_yaml_expanded_nodes)
return ret
[docs]
@staticmethod
def save(
config: Any, f: Union[str, pathlib.Path, IO[Any]], resolve: bool = False
) -> None:
"""
Save as configuration object to a file
:param config: omegaconf.Config object (DictConfig or ListConfig).
:param f: filename or file object
:param resolve: True to save a resolved config (defaults to False)
"""
if is_dataclass(config) or is_attr_class(config):
config = OmegaConf.create(config)
data = OmegaConf.to_yaml(config, resolve=resolve)
if isinstance(f, (str, pathlib.Path)):
with io.open(os.path.abspath(f), "w", encoding="utf-8") as file:
file.write(data)
elif hasattr(f, "write"):
f.write(data)
f.flush()
else:
raise TypeError("Unexpected file type")
[docs]
@staticmethod
def from_cli(args_list: Optional[List[str]] = None) -> DictConfig:
"""
Create a config from command-line arguments (``sys.argv[1:]`` by default).
Each argument must be a dotlist-style string such as ``"foo.bar=1"``.
:param args_list: Explicit list of dotlist strings; defaults to ``sys.argv[1:]``.
:return: A ``DictConfig`` built from the arguments.
"""
if args_list is None:
# Skip program name
args_list = sys.argv[1:]
return OmegaConf.from_dotlist(args_list)
[docs]
@staticmethod
def from_dotlist(dotlist: List[str]) -> DictConfig:
r"""
Creates a config from a list of dotlist-style strings (``"key=value"`` pairs).
Each entry is split on the first *unescaped* ``=``. Everything before
it is the key path; everything after it is the value. Key paths follow
the same dot/bracket syntax as :meth:`select` and :meth:`update`.
**Backslash escaping in keys:**
Use a backslash to include a literal special character in a key name.
The escapable characters are ``.``, ``[``, ``]``, and ``=``.
- ``r"a\.b=1"`` — key is ``"a.b"`` (dot is part of the key)
- ``r"a\=b=1"`` — key is ``"a=b"`` (``=`` is part of the key;
the second ``=`` is the key/value separator)
Values may contain ``=`` freely; only the first unescaped ``=`` separates
key from value (e.g. ``"url=http://x?a=1"`` gives key ``url``,
value ``http://x?a=1``).
**CLI / shell note:** When using :meth:`from_cli`, arguments pass through
the shell before reaching Python. A single backslash in a shell argument
is usually consumed by the shell, so you must double it (``\\``) or
quote the argument (``'a\.b=1'``) to preserve it.
:param dotlist: A list of dotlist-style strings, e.g. ``["foo.bar=1", "baz=qux"]``.
:return: A ``DictConfig`` object created from the dotlist.
"""
conf = OmegaConf.create()
conf.merge_with_dotlist(dotlist)
return conf
[docs]
@staticmethod
def merge(
*configs: Union[
DictConfig,
ListConfig,
Dict[DictKeyType, Any],
List[Any],
Tuple[Any, ...],
Any,
],
list_merge_mode: ListMergeMode = ListMergeMode.REPLACE,
) -> Union[ListConfig, DictConfig]:
"""
Merge a list of previously created configs into a single one
Note for maintainers: changes to merge behavior should also consider
whether OmegaConf.unsafe_merge() needs the same coverage.
:param configs: Input configs
:param list_merge_mode: Behavior for merging lists
REPLACE: Replaces the target list with the new one (default)
EXTEND: Extends the target list with the new one
EXTEND_UNIQUE: Extends the target list items with items not present in it
hint: use `from omegaconf import ListMergeMode` to access the merge mode
:return: the merged config object.
"""
assert len(configs) > 0
target = copy.deepcopy(configs[0])
target = _ensure_container(target)
assert isinstance(target, (DictConfig, ListConfig))
target._merge_with(
*configs[1:],
list_merge_mode=list_merge_mode,
_allow_readonly_target=True,
)
return target
[docs]
@staticmethod
def unsafe_merge(
*configs: Union[
DictConfig,
ListConfig,
Dict[DictKeyType, Any],
List[Any],
Tuple[Any, ...],
Any,
],
list_merge_mode: ListMergeMode = ListMergeMode.REPLACE,
) -> Union[ListConfig, DictConfig]:
"""
Merge a list of previously created configs into a single one
This is much faster than OmegaConf.merge() as the input configs are not copied.
However, the input configs must not be used after this operation as will become inconsistent.
:param configs: Input configs
:param list_merge_mode: Behavior for merging lists
REPLACE: Replaces the target list with the new one (default)
EXTEND: Extends the target list with the new one
EXTEND_UNIQUE: Extends the target list items with items not present in it
hint: use `from omegaconf import ListMergeMode` to access the merge mode
:return: the merged config object.
"""
assert len(configs) > 0
target = configs[0]
target = _ensure_container(target)
assert isinstance(target, (DictConfig, ListConfig))
with flag_override(
target, ["readonly", "no_deepcopy_set_nodes"], [False, True]
):
target._merge_with(
*configs[1:],
list_merge_mode=list_merge_mode,
_allow_readonly_target=True,
)
turned_readonly = target._get_flag("readonly") is True
if turned_readonly:
OmegaConf.set_readonly(target, True)
return target
[docs]
@staticmethod
def register_resolver(
name: str,
resolver: Resolver,
*,
replace: bool = False,
use_cache: bool = False,
) -> None:
"""
Register a resolver.
:param name: Name of the resolver.
:param resolver: Callable whose arguments are provided in the interpolation,
e.g., with ${foo:x,0,${y.z}} these arguments are respectively "x" (str),
0 (int) and the value of ``y.z``.
:param replace: If set to ``False`` (default), then a ``ValueError`` is raised if
an existing resolver has already been registered with the same name.
If set to ``True``, then the new resolver replaces the previous one.
NOTE: The cache on existing config objects is not affected, use
``OmegaConf.clear_cache(cfg)`` to clear it.
:param use_cache: Whether the resolver's outputs should be cached. The cache is
based only on the string literals representing the resolver arguments, e.g.,
${foo:${bar}} will always return the same value regardless of the value of
``bar`` if the cache is enabled for ``foo``.
"""
if not callable(resolver):
raise TypeError("resolver must be callable")
if not name:
raise ValueError("cannot use an empty resolver name")
if not replace and OmegaConf.has_resolver(name):
raise ValueError(f"resolver '{name}' is already registered")
try:
sig: Optional[inspect.Signature] = inspect.signature(resolver)
except ValueError:
sig = None
def _should_pass(special: str) -> bool:
ret = sig is not None and special in sig.parameters
if ret and use_cache:
raise ValueError(
f"use_cache=True is incompatible with functions that receive the {special}"
)
return ret
pass_parent = _should_pass("_parent_")
pass_node = _should_pass("_node_")
pass_root = _should_pass("_root_")
def resolver_wrapper(
config: BaseContainer,
parent: Container,
node: Node,
args: Tuple[Any, ...],
args_str: Tuple[str, ...],
) -> Any:
if use_cache:
cache = OmegaConf.get_cache(config)[name]
try:
return cache[args_str]
except KeyError:
ret = resolver(*args)
cache[args_str] = ret
return ret
# Call resolver.
kwargs: Dict[str, Node] = {}
if pass_parent:
kwargs["_parent_"] = parent
if pass_node:
kwargs["_node_"] = node
if pass_root:
kwargs["_root_"] = config
ret = resolver(*args, **kwargs)
return ret
# noinspection PyProtectedMember
BaseContainer._resolvers[name] = resolver_wrapper
@staticmethod
def register_new_resolver(
name: str,
resolver: Resolver,
*,
replace: bool = False,
use_cache: bool = False,
) -> None:
warnings.warn(
dedent("""\
register_new_resolver() is deprecated and will be removed in a future release.
Use register_resolver() instead.
See https://github.com/omry/omegaconf/issues/426 for migration instructions.
"""),
stacklevel=2,
)
return OmegaConf.register_resolver(
name, resolver, replace=replace, use_cache=use_cache
)
@staticmethod
def legacy_register_resolver(name: str, resolver: Resolver) -> None:
warnings.warn(
dedent("""\
legacy_register_resolver() is deprecated and will be removed in a future release.
Use register_resolver() instead.
See https://github.com/omry/omegaconf/issues/426 for migration instructions.
"""),
stacklevel=2,
)
assert callable(resolver), "resolver must be callable"
# noinspection PyProtectedMember
assert (
name not in BaseContainer._resolvers
), f"resolver '{name}' is already registered"
def resolver_wrapper(
config: BaseContainer,
parent: BaseContainer,
node: Node,
args: Tuple[Any, ...],
args_str: Tuple[str, ...],
) -> Any:
cache = OmegaConf.get_cache(config)[name]
# "Un-escape " spaces and commas.
args_unesc = [x.replace(r"\ ", " ").replace(r"\,", ",") for x in args_str]
# Nested interpolations behave in a potentially surprising way with
# legacy resolvers (they remain as strings, e.g., "${foo}"). If any
# input looks like an interpolation we thus raise an exception.
try:
bad_arg = next(i for i in args_unesc if "${" in i)
except StopIteration:
pass
else:
raise ValueError(
f"Resolver '{name}' was called with argument '{bad_arg}' that appears "
f"to be an interpolation. Nested interpolations are not supported for "
f"resolvers registered with `legacy_register_resolver()`, please use "
f"`register_resolver()` instead (see "
f"https://github.com/omry/omegaconf/issues/426 for migration instructions)." # noqa: E231
)
key = args_str
val = cache[key] if key in cache else resolver(*args_unesc)
cache[key] = val
return val
# noinspection PyProtectedMember
BaseContainer._resolvers[name] = resolver_wrapper
[docs]
@classmethod
def has_resolver(cls, name: str) -> bool:
"""
Return ``True`` if a resolver with the given name is registered.
:param name: Resolver name to check.
:return: ``True`` if registered, ``False`` otherwise.
"""
return cls._get_resolver(name) is not None
# noinspection PyProtectedMember
[docs]
@staticmethod
def clear_resolvers() -> None:
"""
Clear(remove) all OmegaConf resolvers, then re-register OmegaConf's default resolvers.
"""
BaseContainer._resolvers = {}
register_default_resolvers()
[docs]
@classmethod
def clear_resolver(cls, name: str) -> bool:
"""
Clear(remove) any resolver only if it exists.
Returns a bool: True if resolver is removed and False if not removed.
.. warning:
This method can remove default resolvers as well.
:param name: Name of the resolver.
:return: A bool (``True`` if resolver is removed, ``False`` if not found before removing).
"""
if cls.has_resolver(name):
BaseContainer._resolvers.pop(name)
return True
else:
# return False if resolver does not exist
return False
[docs]
@staticmethod
def get_cache(conf: BaseContainer) -> Dict[str, Any]:
"""
Return the resolver cache for ``conf``.
:param conf: An OmegaConf container.
:return: The resolver cache dict (resolver name -> cached values).
"""
return conf._metadata.resolver_cache
[docs]
@staticmethod
def set_cache(conf: BaseContainer, cache: Dict[str, Any]) -> None:
"""
Replace the resolver cache for ``conf`` with a deep copy of ``cache``.
:param conf: An OmegaConf container.
:param cache: New cache dict to install (will be deep-copied).
"""
conf._metadata.resolver_cache = copy.deepcopy(cache)
[docs]
@staticmethod
def clear_cache(conf: BaseContainer) -> None:
"""
Clear the resolver cache for ``conf``.
:param conf: An OmegaConf container.
"""
OmegaConf.set_cache(conf, defaultdict(dict, {}))
[docs]
@staticmethod
def copy_cache(from_config: BaseContainer, to_config: BaseContainer) -> None:
"""
Copy the resolver cache from one config to another.
:param from_config: Source container whose cache is copied.
:param to_config: Destination container that receives the cache copy.
"""
OmegaConf.set_cache(to_config, OmegaConf.get_cache(from_config))
[docs]
@staticmethod
def set_readonly(conf: Node, value: Optional[bool]) -> None:
"""
Set the read-only flag on ``conf``.
:param conf: An OmegaConf node.
:param value: ``True`` to make read-only, ``False`` to make writable,
``None`` to inherit from the parent.
"""
# noinspection PyProtectedMember
conf._set_flag("readonly", value)
[docs]
@staticmethod
def is_readonly(conf: Node) -> Optional[bool]:
"""
Return the effective read-only flag of ``conf``.
:param conf: An OmegaConf node.
:return: ``True`` if read-only, ``False`` if writable, ``None`` if not set
(inherits from parent).
"""
# noinspection PyProtectedMember
return conf._get_flag("readonly")
[docs]
@staticmethod
def set_struct(conf: Container, value: Optional[bool]) -> None:
"""
Set the struct flag on ``conf``.
When struct mode is enabled, accessing or setting keys that do not exist in the
config raises an exception.
:param conf: An OmegaConf container.
:param value: ``True`` to enable struct mode, ``False`` to disable it,
``None`` to inherit from the parent.
"""
# noinspection PyProtectedMember
conf._set_flag("struct", value)
[docs]
@staticmethod
def is_struct(conf: Container) -> Optional[bool]:
"""
Return the effective struct flag of ``conf``.
:param conf: An OmegaConf container.
:return: ``True`` if struct mode is on, ``False`` if off, ``None`` if not set
(inherits from parent).
"""
# noinspection PyProtectedMember
return conf._get_flag("struct")
[docs]
@staticmethod
def masked_copy(conf: DictConfig, keys: Union[str, List[str]]) -> DictConfig:
"""
Create a masked copy of of this config that contains a subset of the keys
:param conf: DictConfig object
:param keys: keys to preserve in the copy
:return: The masked ``DictConfig`` object.
"""
from .dictconfig import DictConfig
if not isinstance(conf, DictConfig):
raise ValueError("masked_copy is only supported for DictConfig")
if isinstance(keys, str):
keys = [keys]
content = {key: value for key, value in conf.items_ex(resolve=False, keys=keys)}
return DictConfig(content=content)
[docs]
@staticmethod
def to_container(
cfg: Any,
*,
resolve: bool = False,
throw_on_missing: bool = False,
enum_to_str: bool = False,
structured_config_mode: SCMode = SCMode.DICT,
) -> Union[Dict[DictKeyType, Any], List[Any], None, str, Any]:
"""
Recursively converts an OmegaConf config to a primitive container (dict or list).
:param cfg: the config to convert
:param resolve: True to resolve all values
:param throw_on_missing: When True, raise MissingMandatoryValue if any missing values are present.
When False (the default), replace missing values with the string "???" in the output container.
:param enum_to_str: True to convert Enum keys and values to strings
:param structured_config_mode: Specify how Structured Configs (DictConfigs backed by a dataclass) are handled.
- By default (``structured_config_mode=SCMode.DICT``) structured configs are converted to plain dicts.
- If ``structured_config_mode=SCMode.DICT_CONFIG``, structured config nodes will remain as DictConfig.
- If ``structured_config_mode=SCMode.INSTANTIATE``, this function will instantiate structured configs
(DictConfigs backed by a dataclass), by creating an instance of the underlying dataclass.
See also OmegaConf.to_object.
:return: A dict or a list representing this config as a primitive container.
"""
if not OmegaConf.is_config(cfg):
raise ValueError(
f"Input cfg is not an OmegaConf config object ({type_str(type(cfg))})"
)
return BaseContainer._to_content(
cfg,
resolve=resolve,
throw_on_missing=throw_on_missing,
enum_to_str=enum_to_str,
structured_config_mode=structured_config_mode,
)
[docs]
@staticmethod
def structural_equality(cfg1: Any, cfg2: Any) -> bool:
"""
Compare two configs by their unresolved container structure.
This is equivalent to converting both configs with
``OmegaConf.to_container(resolve=False, throw_on_missing=False)`` and
comparing the resulting containers. Interpolations and custom resolver
expressions are compared as their raw strings and are not resolved.
Missing values do not raise.
:param cfg1: First OmegaConf config to compare.
:param cfg2: Second OmegaConf config to compare.
:return: ``True`` if both configs have the same unresolved structure.
"""
return OmegaConf.to_container(
cfg1, resolve=False, throw_on_missing=False
) == OmegaConf.to_container(cfg2, resolve=False, throw_on_missing=False)
[docs]
@staticmethod
def to_object(cfg: Any) -> Union[Dict[DictKeyType, Any], List[Any], None, str, Any]:
"""
Recursively converts an OmegaConf config to a primitive container (dict or list).
Any DictConfig objects backed by dataclasses or attrs classes are instantiated
as instances of those backing classes.
This is an alias for OmegaConf.to_container(..., resolve=True, throw_on_missing=True,
structured_config_mode=SCMode.INSTANTIATE)
:param cfg: the config to convert
:return: A dict or a list or dataclass representing this config.
"""
return OmegaConf.to_container(
cfg=cfg,
resolve=True,
throw_on_missing=True,
enum_to_str=False,
structured_config_mode=SCMode.INSTANTIATE,
)
[docs]
@staticmethod
def is_missing(cfg: Any, key: DictKeyType) -> bool:
"""
Return ``True`` if ``cfg[key]`` is set to the mandatory-missing sentinel ``???``.
:param cfg: An OmegaConf container.
:param key: Key (str for DictConfig, int for ListConfig) to check.
:return: ``True`` if the value is missing, ``False`` otherwise.
"""
assert isinstance(cfg, Container)
try:
node = cfg._get_child(key)
if node is None:
return False
assert isinstance(node, Node)
return node._is_missing()
except (UnsupportedInterpolationType, KeyError, AttributeError):
return False
[docs]
@staticmethod
def is_interpolation(node: Any, key: Optional[Union[int, str]] = None) -> bool:
"""
Return ``True`` if the target node is an interpolation (e.g. ``${foo.bar}``).
If ``key`` is provided, checks ``node[key]``; otherwise checks ``node`` itself.
:param node: An OmegaConf node, or a container when ``key`` is given.
:param key: Optional key within ``node`` to inspect.
:return: ``True`` if the value is an interpolation, ``False`` otherwise.
"""
if key is not None:
assert isinstance(node, Container)
target = node._get_child(key)
else:
target = node
if target is not None:
assert isinstance(target, Node)
return target._is_interpolation()
return False
[docs]
@staticmethod
def is_list(obj: Any) -> bool:
"""
Return ``True`` if ``obj`` is an OmegaConf ``ListConfig``.
:param obj: Object to test.
:return: ``True`` if ``obj`` is a ``ListConfig``, ``False`` otherwise.
"""
from . import ListConfig
return isinstance(obj, ListConfig)
[docs]
@staticmethod
def is_dict(obj: Any) -> bool:
"""
Return ``True`` if ``obj`` is an OmegaConf ``DictConfig``.
:param obj: Object to test.
:return: ``True`` if ``obj`` is a ``DictConfig``, ``False`` otherwise.
"""
from . import DictConfig
return isinstance(obj, DictConfig)
[docs]
@staticmethod
def is_config(obj: Any) -> bool:
"""
Return ``True`` if ``obj`` is an OmegaConf config (``DictConfig`` or ``ListConfig``).
:param obj: Object to test.
:return: ``True`` if ``obj`` is an OmegaConf container, ``False`` otherwise.
"""
from . import Container
return isinstance(obj, Container)
[docs]
@staticmethod
def get_type(obj: Any, key: Optional[str] = None) -> Optional[Type[Any]]:
"""
Return the type of ``obj``, or of ``obj[key]`` when ``key`` is provided.
For structured configs this is the underlying dataclass or attrs class.
For plain containers it is ``dict`` or ``list``.
:param obj: An OmegaConf node or container.
:param key: Optional key within ``obj`` to inspect.
:return: The Python type, or ``None`` if not determinable.
"""
if key is not None:
c = obj._get_child(key)
else:
c = obj
return OmegaConf._get_obj_type(c)
[docs]
@staticmethod
def can_select(
cfg: Container,
key: str,
*,
throw_on_resolution_failure: bool = True,
throw_on_missing: bool = False,
) -> bool:
r"""
Return ``True`` if ``OmegaConf.select()`` can select a value from a config.
This uses the same key path syntax and behavior flag names as
``select()`` for convenience, but ``can_select()`` does not raise for
select failures. It returns ``False`` instead of raising or returning a
default when the path cannot be selected. A selected ``None`` value
counts as selectable.
:param cfg: Config node to select from
:param key: Key path to select (dot/bracket notation, backslash-escapable)
:param throw_on_resolution_failure: Treat an interpolation resolution error
as not selectable
:param throw_on_missing: Treat selecting a missing key (with the value
'???') as not selectable
:return: ``True`` if the key can be selected, otherwise ``False``.
"""
from ._impl import select_value
default = object()
try:
return (
select_value(
cfg=cfg,
key=key,
default=default,
throw_on_resolution_failure=throw_on_resolution_failure,
throw_on_missing=throw_on_missing,
)
is not default
)
except Exception:
return False
[docs]
@staticmethod
def select(
cfg: Container,
key: str,
*,
default: Any = _DEFAULT_MARKER_,
throw_on_resolution_failure: bool = True,
throw_on_missing: bool = False,
) -> Any:
r"""
Select a value from a config using a key path.
The key path uses dot notation (``"a.b.c"``) or bracket notation
(``"a[b][c]"``), or a mix of both.
**Keys containing special characters** (``.``, ``[``, ``]``, ``=``)
can be expressed by escaping them with a backslash:
- ``r"a\.b"`` — selects the key ``"a.b"`` (single key with a literal dot)
- ``r"a\[0\]"`` — selects the key ``"a[0]"``
- ``r"a\=b"`` — selects the key ``"a=b"``
A backslash before any other character passes through unchanged
(``r"a\b"`` selects the key ``"a\\b"`` — a backslash followed by ``b``).
:param cfg: Config node to select from
:param key: Key path to select (dot/bracket notation, backslash-escapable)
:param default: Default value to return if key is not found
:param throw_on_resolution_failure: Raise an exception if an interpolation
resolution error occurs, otherwise return None
:param throw_on_missing: Raise an exception if an attempt to select a missing key (with the value '???')
is made, otherwise return None
:return: selected value or None if not found.
"""
from ._impl import select_value
try:
return select_value(
cfg=cfg,
key=key,
default=default,
throw_on_resolution_failure=throw_on_resolution_failure,
throw_on_missing=throw_on_missing,
)
except Exception as e:
format_and_raise(node=cfg, key=key, value=None, cause=e, msg=str(e))
[docs]
@staticmethod
def update(
cfg: Container,
key: str,
value: Any = None,
*,
merge: bool = True,
force_add: bool = False,
) -> None:
r"""
Update a value in a config using a key path.
The key path uses dot notation (``"a.b.c"``) or bracket notation
(``"a[b][c]"``), or a mix of both.
**Keys containing special characters** (``.``, ``[``, ``]``, ``=``)
can be expressed by escaping them with a backslash:
- ``r"a\.b"`` — targets the key ``"a.b"`` (single key with a literal dot)
- ``r"a\[0\]"`` — targets the key ``"a[0]"``
- ``r"a\=b"`` — targets the key ``"a=b"``
:param cfg: input config to update
:param key: key path to update (dot/bracket notation, backslash-escapable)
:param value: value to set, if value if a list or a dict it will be merged or set
depending on merge_config_values
:param merge: If value is a dict or a list, True (default) to merge
into the destination, False to replace the destination.
:param force_add: insert the entire path regardless of Struct flag or Structured Config nodes.
"""
split = split_key(key)
root = cfg
for i in range(len(split) - 1):
k = split[i]
# if next_root is a primitive (string, int etc) replace it with an empty map
next_root, key_ = _select_one(root, k, throw_on_missing=False)
if isinstance(next_root, Container) and next_root._is_none():
raise ConfigTypeError(
f"Cannot set '{key}' because"
f" '{root._get_full_key(key_)}' is None"
)
if not isinstance(next_root, Container):
if force_add:
with flag_override(root, "struct", False):
root[key_] = {}
else:
root[key_] = {}
root = root[key_]
last = split[-1]
assert isinstance(
root, Container
), f"Unexpected type for root: {type(root).__name__}"
last_key: Union[str, int] = last
if isinstance(root, ListConfig):
last_key = int(last)
ctx = flag_override(root, "struct", False) if force_add else nullcontext()
with ctx:
if merge and (OmegaConf.is_config(value) or is_primitive_container(value)):
assert isinstance(root, BaseContainer)
node = root._get_child(last_key)
if OmegaConf.is_config(node):
assert isinstance(node, BaseContainer)
node.merge_with(value)
return
if OmegaConf.is_dict(root):
assert isinstance(last_key, str)
root.__setattr__(last_key, value)
elif OmegaConf.is_list(root):
assert isinstance(last_key, int)
root.__setitem__(last_key, value)
else:
assert False
[docs]
@staticmethod
def to_yaml(
cfg: Any,
*,
resolve: bool = False,
sort_keys: bool = False,
default_flow_style: Optional[bool] = False,
) -> str:
"""
returns a yaml dump of this config object.
:param cfg: Config object, Structured Config type or instance
:param resolve: if True, will return a string with the interpolations resolved, otherwise
interpolations are preserved
:param sort_keys: If True, will print dict keys in sorted order. default False.
:param default_flow_style: PyYAML default_flow_style setting. default False.
:return: A string containing the yaml representation.
"""
cfg = _ensure_container(cfg)
container = OmegaConf.to_container(cfg, resolve=resolve, enum_to_str=True)
return yaml.dump( # type: ignore
container,
default_flow_style=default_flow_style,
allow_unicode=True,
sort_keys=sort_keys,
Dumper=get_omega_conf_dumper(),
)
[docs]
@staticmethod
def resolve(cfg: Container) -> None:
"""
Resolves all interpolations in the given config object in-place.
This function works correctly for configs that use only node interpolations
(``${key}``) with no custom resolvers. When custom resolvers are involved,
results may depend on the depth-first, key insertion order traversal, because
custom resolvers can do anything — they may be stateful, have side effects, or
return different values on each call — and this function has no way to account
for that.
:param cfg: An OmegaConf container (DictConfig or ListConfig).
:raises ValueError: If the input object is not an OmegaConf container.
"""
import omegaconf._impl
if not OmegaConf.is_config(cfg):
# Since this function is mutating the input object in-place, it doesn't make sense to
# auto-convert the input object to an OmegaConf container
raise ValueError(
f"Invalid config type ({type(cfg).__name__}), expected an OmegaConf Container"
)
omegaconf._impl._resolve(cfg)
[docs]
@staticmethod
def missing_keys(cfg: Any, *, resolve_custom_resolvers: bool = False) -> Set[str]:
"""
Returns a set of missing keys in a dotlist style.
Node interpolations that dereference missing values are reported as missing
keys, whether they are the full value or part of a string.
:param cfg: An ``OmegaConf.Container``,
or a convertible object via ``OmegaConf.create`` (dict, list, ...).
:param resolve_custom_resolvers: If ``True``, custom resolver
interpolations are resolved and reported as missing when they
dereference missing values. If ``False`` (the default),
custom resolver interpolations are not resolved and are not
reported as missing keys.
:return: set of strings of the missing keys.
:raises ValueError: On input not representing a config.
"""
cfg = _ensure_container(cfg)
missings: Set[str] = set()
def contains_custom_resolver_interpolation(node: Node) -> bool:
value = node._value()
assert isinstance(value, str)
if _CUSTOM_RESOLVER_INTERPOLATION_PATTERN.search(value) is None:
return False
from .grammar_parser import OmegaConfGrammarParser, parse
def has_resolver_interpolation(ctx: Any) -> bool:
if isinstance(ctx, OmegaConfGrammarParser.InterpolationResolverContext):
return True
get_children = getattr(ctx, "getChildren", None)
if get_children is None:
return False
return any(
has_resolver_interpolation(child) for child in get_children()
)
return has_resolver_interpolation(parse(value))
def is_missing_value_error(exc: BaseException) -> bool:
current: Optional[BaseException] = exc
while current is not None:
if isinstance(
current,
(InterpolationToMissingValueError, MissingMandatoryValue),
):
return True
current = (
current.__cause__
if current.__cause__ is not None
else current.__context__
)
return False
def gather(_cfg: Container) -> None:
itr: Iterable[Any]
if isinstance(_cfg, ListConfig):
itr = range(len(_cfg))
else:
itr = _cfg
for key in itr:
node = _cfg._get_child(key)
assert isinstance(node, Node)
if node._is_missing():
missings.add(_cfg._get_full_key(key))
elif (
not resolve_custom_resolvers
and node._is_interpolation()
and contains_custom_resolver_interpolation(node)
):
continue
else:
try:
value = _cfg[key]
except InterpolationResolutionError as exc:
if is_missing_value_error(exc):
missings.add(_cfg._get_full_key(key))
else:
raise
else:
if OmegaConf.is_config(value):
gather(value)
gather(cfg)
return missings
# === private === #
@staticmethod
def _create_impl( # noqa F811
obj: Any = _DEFAULT_MARKER_,
parent: Optional[BaseContainer] = None,
flags: Optional[Dict[str, bool]] = None,
*,
max_yaml_expanded_nodes: Optional[int] = _DEFAULT_MAX_YAML_EXPANDED_NODES,
) -> Optional[Union[DictConfig, ListConfig]]:
try:
from .dictconfig import DictConfig
from .listconfig import ListConfig
if obj is _DEFAULT_MARKER_:
obj = {}
elif obj is None:
return None
if isinstance(obj, str):
obj = yaml.load(
obj,
Loader=get_yaml_loader(
max_yaml_expanded_nodes=max_yaml_expanded_nodes
),
)
if obj is None:
return OmegaConf.create({}, parent=parent, flags=flags)
elif isinstance(obj, str):
return OmegaConf.create({obj: None}, parent=parent, flags=flags)
else:
assert isinstance(obj, (list, dict))
return OmegaConf.create(obj, parent=parent, flags=flags)
else:
if (
is_primitive_dict(obj)
or OmegaConf.is_dict(obj)
or is_structured_config(obj)
or obj is None
):
if isinstance(obj, DictConfig):
return DictConfig(
content=obj,
parent=parent,
ref_type=obj._metadata.ref_type,
is_optional=obj._metadata.optional,
key_type=obj._metadata.key_type,
element_type=obj._metadata.element_type,
flags=flags,
)
else:
obj_type = OmegaConf.get_type(obj)
key_type, element_type = get_dict_key_value_types(obj_type)
return DictConfig(
content=obj,
parent=parent,
key_type=key_type,
element_type=element_type,
flags=flags,
)
elif is_primitive_list(obj) or OmegaConf.is_list(obj):
if isinstance(obj, ListConfig):
return ListConfig(
content=obj,
parent=parent,
element_type=obj._metadata.element_type,
ref_type=obj._metadata.ref_type,
is_optional=obj._metadata.optional,
flags=flags,
)
else:
obj_type = OmegaConf.get_type(obj)
element_type = get_list_element_type(obj_type)
return ListConfig(
content=obj,
parent=parent,
element_type=element_type,
ref_type=Any,
is_optional=True,
flags=flags,
)
else:
if isinstance(obj, type):
raise ValidationError(
f"Input class '{obj.__name__}' is not a structured config. "
"did you forget to decorate it as a dataclass?"
)
else:
raise ValidationError(
f"Object of unsupported type: '{type(obj).__name__}'"
)
except OmegaConfBaseException as e:
format_and_raise(node=None, key=None, value=None, msg=str(e), cause=e)
assert False
@staticmethod
def _get_obj_type(c: Any) -> Optional[Type[Any]]:
if is_structured_config(c):
return get_type_of(c)
elif c is None:
return None
elif isinstance(c, DictConfig):
if c._is_none():
return None
elif c._is_missing():
return None
else:
if is_structured_config(c._metadata.object_type):
return c._metadata.object_type
else:
return dict
elif isinstance(c, ListConfig):
return list
elif isinstance(c, ValueNode):
return type(c._value())
elif isinstance(c, UnionNode):
return type(_get_value(c))
elif isinstance(c, dict):
return dict
elif isinstance(c, (list, tuple)):
return list
else:
return get_type_of(c)
@staticmethod
def _get_resolver(
name: str,
) -> Optional[
Callable[
[Container, Container, Node, Tuple[Any, ...], Tuple[str, ...]],
Any,
]
]:
# noinspection PyProtectedMember
return (
BaseContainer._resolvers[name] if name in BaseContainer._resolvers else None
)
# register all default resolvers
register_default_resolvers()
@contextmanager
def flag_override(
config: Node,
names: Union[List[str], str],
values: Union[List[Optional[bool]], Optional[bool]],
) -> Generator[Node, None, None]:
"""
Context manager that temporarily overrides one or more flags on ``config``.
The original flag values are restored on exit, even if an exception is raised.
:param config: An OmegaConf node whose flags will be overridden.
:param names: Flag name or list of flag names (e.g. ``"readonly"``, ``"struct"``).
:param values: New value or list of values corresponding to ``names``.
:return: Yields ``config`` with the overridden flags.
"""
if isinstance(names, str):
names = [names]
if values is None or isinstance(values, bool):
values = [values]
prev_states = [config._get_node_flag(name) for name in names]
try:
config._set_flag(names, values)
yield config
finally:
config._set_flag(names, prev_states)
@contextmanager
def read_write(config: Node) -> Generator[Node, None, None]:
"""
Context manager that temporarily makes ``config`` writable.
The original read-only state is restored on exit, even if an exception is raised.
:param config: An OmegaConf node.
:return: Yields ``config`` in a writable state.
"""
prev_state = config._get_node_flag("readonly")
try:
OmegaConf.set_readonly(config, False)
yield config
finally:
OmegaConf.set_readonly(config, prev_state)
@contextmanager
def open_dict(config: Container) -> Generator[Container, None, None]:
"""
Context manager that temporarily disables struct mode on ``config``.
While active, new keys can be added freely. The original struct state is restored
on exit, even if an exception is raised.
:param config: An OmegaConf container.
:return: Yields ``config`` with struct mode disabled.
"""
prev_state = config._get_node_flag("struct")
try:
OmegaConf.set_struct(config, False)
yield config
finally:
OmegaConf.set_struct(config, prev_state)
# === private === #
def _node_wrap(
parent: Optional[Box],
is_optional: bool,
value: Any,
key: Any,
ref_type: Any = Any,
) -> Node:
node: Node
if is_dict_annotation(ref_type) or (is_primitive_dict(value) and ref_type is Any):
key_type, element_type = get_dict_key_value_types(ref_type)
node = DictConfig(
content=value,
key=key,
parent=parent,
ref_type=ref_type,
is_optional=is_optional,
key_type=key_type,
element_type=element_type,
)
elif (is_list_annotation(ref_type) or is_tuple_annotation(ref_type)) or (
type(value) in (list, tuple) and ref_type is Any
):
element_type = get_list_element_type(ref_type)
node = ListConfig(
content=value,
key=key,
parent=parent,
is_optional=is_optional,
element_type=element_type,
ref_type=ref_type,
)
elif is_structured_config(ref_type) or is_structured_config(value):
key_type, element_type = get_dict_key_value_types(value)
node = DictConfig(
ref_type=ref_type,
is_optional=is_optional,
content=value,
key=key,
parent=parent,
key_type=key_type,
element_type=element_type,
)
elif is_union_annotation(ref_type):
node = UnionNode(
content=value,
ref_type=ref_type,
is_optional=is_optional,
key=key,
parent=parent,
)
elif is_literal_annotation(ref_type):
node = LiteralNode(
ref_type=ref_type,
value=value,
key=key,
parent=parent,
is_optional=is_optional,
)
elif ref_type == Any or ref_type is None:
node = AnyNode(value=value, key=key, parent=parent)
elif isinstance(ref_type, type) and issubclass(ref_type, Enum):
node = EnumNode(
enum_type=ref_type,
value=value,
key=key,
parent=parent,
is_optional=is_optional,
)
elif ref_type == int:
node = IntegerNode(value=value, key=key, parent=parent, is_optional=is_optional)
elif ref_type == float:
node = FloatNode(value=value, key=key, parent=parent, is_optional=is_optional)
elif ref_type == bool:
node = BooleanNode(value=value, key=key, parent=parent, is_optional=is_optional)
elif ref_type == str:
node = StringNode(value=value, key=key, parent=parent, is_optional=is_optional)
elif ref_type == bytes:
node = BytesNode(value=value, key=key, parent=parent, is_optional=is_optional)
elif ref_type == pathlib.Path:
node = PathNode(value=value, key=key, parent=parent, is_optional=is_optional)
else:
if parent is not None and parent._get_flag("allow_objects") is True:
if type(value) in (list, tuple):
node = ListConfig(
content=value,
key=key,
parent=parent,
ref_type=ref_type,
is_optional=is_optional,
)
elif is_primitive_dict(value):
node = DictConfig(
content=value,
key=key,
parent=parent,
ref_type=ref_type,
is_optional=is_optional,
)
else:
node = AnyNode(value=value, key=key, parent=parent)
else:
raise ValidationError(f"Unexpected type annotation: {type_str(ref_type)}")
return node
def _maybe_wrap(
ref_type: Any,
key: Any,
value: Any,
is_optional: bool,
parent: Optional[BaseContainer],
) -> Node:
# if already a node, update key and parent and return as is.
# NOTE: that this mutate the input node!
if isinstance(value, Node):
value._set_key(key)
value._set_parent(parent)
return value
else:
return _node_wrap(
ref_type=ref_type,
parent=parent,
is_optional=is_optional,
value=value,
key=key,
)
def _select_one(
c: Container, key: str, throw_on_missing: bool, throw_on_type_error: bool = True
) -> Tuple[Optional[Node], Union[str, int]]:
from .dictconfig import DictConfig
from .listconfig import ListConfig
ret_key: Union[str, int] = key
assert isinstance(c, Container), f"Unexpected type: {c}"
if c._is_none():
return None, ret_key
if isinstance(c, DictConfig):
assert isinstance(ret_key, str)
val = c._get_child(ret_key, validate_access=False)
elif isinstance(c, ListConfig):
assert isinstance(ret_key, str)
if not is_int(ret_key):
if throw_on_type_error:
raise TypeError(
f"Index '{ret_key}' ({type(ret_key).__name__}) is not an int"
)
else:
val = None
else:
ret_key = int(ret_key)
if ret_key < 0:
ret_key += len(c)
if ret_key < 0 or ret_key + 1 > len(c):
val = None
else:
val = c._get_child(ret_key)
else:
assert False
if val is not None:
assert isinstance(val, Node)
if val._is_missing():
if throw_on_missing:
raise MissingMandatoryValue(
f"Missing mandatory value: {c._get_full_key(ret_key)}"
)
else:
return val, ret_key
assert val is None or isinstance(val, Node)
return val, ret_key