Skip to content

confit.config

Config

Bases: dict

The configuration system consists of a supercharged dict, the Config class, that can be used to read and write to cfg files, interpolate variables and instantiate components through the registry with some special @factory keys. A cfg file can be used directly as an input to a CLI-decorated function.

Source code in confit/config.py
 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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
class Config(dict):
    """
    The configuration system consists of a supercharged dict, the `Config` class,
    that can be used to read and write to `cfg` files, interpolate variables and
    instantiate components through the registry with some special `@factory` keys.
    A cfg file can be used directly as an input to a CLI-decorated function.
    """

    def __init__(self, *args: Any, **kwargs: Any):
        """
        A new config object can be instantiated either from a dict as a positional
        argument, or from keyword arguments. Only one of these two options can be
        used at a time.

        Parameters
        ----------
        args: Any
        kwargs: Any
        """
        if len(args) == 1 and isinstance(args[0], dict):
            assert len(kwargs) == 0
            kwargs = args[0]
        super().__init__(**kwargs)

    @classmethod
    def from_cfg_str(cls, s: str, resolve: bool = False, registry: Any = None) -> Any:
        """
        Load a config object from a config string

        Parameters
        ----------
        s: Union[str, Path]
            The cfg config string
        resolve
            Whether to resolve sections with '@' keys
        registry
            Optional registry to resolve from.
            If None, the default registry will be used.

        Returns
        -------
        Config
        """
        parser = ConfigParser(interpolation=None)
        parser.optionxform = str
        parser.read_string(s)

        config = Config()

        for section in parser.sections():
            parts = split_path(section)
            current = config
            for part in parts:
                if part not in current:
                    current[part] = current = Config()
                else:
                    current = current[part]

            current.clear()
            errors = []
            for k, v in parser.items(section):
                path = split_path(k)
                for part in path[:-1]:
                    if part not in current:
                        current[part] = current = Config()
                    else:
                        current = current[part]
                try:
                    current[path[-1]] = loads(v)
                except ValueError as e:
                    errors.append(ErrorWrapper(e, loc=path))

            if errors:
                raise ConfitValidationError(errors=errors)

        if resolve:
            return config.resolve(registry=registry)

        return config

    @classmethod
    def from_yaml_str(cls, s: str, resolve: bool = False, registry: Any = None) -> Any:
        import yaml

        class ConfitYamlLoader(yaml.SafeLoader):
            def construct_object(self, x, deep=False):
                if isinstance(x, yaml.ScalarNode):
                    return loads(self.buffer[x.start_mark.index : x.end_mark.index])
                return super().construct_object(x, deep)

        stream = StringIO(s)
        config = Config(yaml.load(stream, Loader=ConfitYamlLoader))
        if resolve:
            return config.resolve(registry=registry)

        return config

    @classmethod
    def from_disk(
        cls, path: Union[str, Path], resolve: bool = False, registry: Any = None
    ) -> "Config":
        """
        Load a config object from a '.cfg' file

        Parameters
        ----------
        path: Union[str, Path]
            The path to the config object
        resolve
            Whether to resolve mappings with '@' keys
        registry
            Optional registry to resolve from.
            If None, the default registry will be used.

        Returns
        -------
        Config
        """
        s = Path(path).read_text(encoding="utf-8")
        path_str = str(path)
        if path_str.endswith(".yaml") or path_str.endswith(".yml"):
            return cls.from_yaml_str(s, resolve=resolve, registry=registry)
        else:
            return cls.from_cfg_str(s, resolve=resolve, registry=registry)

    from_str = from_cfg_str

    def to_disk(self, path: Union[str, Path]):
        """
        Export a config to the disk (usually to a .cfg file)

        Parameters
        ----------
        path: Union[str, path]
        """
        path_str = str(path)
        if path_str.endswith(".yaml") or path_str.endswith(".yml"):
            s = self.to_yaml_str()
        else:
            s = Config.to_str(self)
        Path(path).write_text(s)

    def serialize(self: Any):
        """
        Try to convert non-serializable objects using the RESOLVED_TO_CONFIG object
        back to their original catalogue + params form

        We try to preserve referential equalities between non dict/list/tuple
        objects by serializing subsequent references to the same object as references
        to its first occurrence in the tree.

        ```python
        a = A()  # serializable object
        cfg = {"a": a, "b": a}
        print(Config.serialize(cfg))
        # Out: {"a": {...}, "b": Reference("a")}
        ```

        Returns
        -------
        Config
        """
        refs = {}

        # Temp memory to avoid objects being garbage collected
        mem = []

        def is_simple(o):
            return o is None or isinstance(o, (str, int, float, bool, Reference))

        def rec(o: Any, path: Loc = ()):
            if id(o) in refs:
                return refs[id(o)]
            if is_simple(o):
                return o
            if isinstance(o, collections.abc.Mapping):
                items = sorted(
                    o.items(),
                    key=lambda x: 1
                    if (
                        is_simple(x[1])
                        or isinstance(x[1], (collections.abc.Mapping, list, tuple))
                    )
                    else 0,
                )
                serialized = {k: rec(v, (*path, k)) for k, v in items}
                serialized = {k: serialized[k] for k in o.keys()}
                mem.append(o)
                refs[id(o)] = Reference(join_path(path))
                if isinstance(o, Config):
                    serialized = Config(serialized)
                return serialized
            if isinstance(o, (list, tuple)):
                mem.append(o)
                refs[id(o)] = Reference(join_path(path))
                return type(o)(rec(v, (*path, i)) for i, v in enumerate(o))
            cfg = None
            try:
                cfg = (cfg or Config()).merge(RESOLVED_TO_CONFIG[o])
            except (KeyError, TypeError):
                pass
            try:
                cfg = (cfg or Config()).merge(o.cfg)
            except (AttributeError, TypeError):
                pass
            if cfg is not None:
                mem.append(o)
                refs[id(o)] = Reference(join_path(path))
                return rec(cfg, path)
            try:
                return pydantic_core.to_jsonable_python(o)
            except Exception:
                raise TypeError(f"Cannot dump {o!r} at {join_path(path)}")

        return rec(self)

    def to_cfg_str(self):
        """
        Export a config to a string in the cfg format
        by serializing it first

        Returns
        -------
        str
        """
        additional_sections = {}

        prepared = flatten_sections(Config.serialize(self))
        prepared.update(flatten_sections(additional_sections))

        parser = ConfigParser(interpolation=None)
        parser.optionxform = str
        for section_name, section in prepared.items():
            parser.add_section(section_name)
            parser[section_name].update(
                {join_path((k,)): dumps(v) for k, v in section.items()}
            )
        s = StringIO()
        parser.write(s)
        return s.getvalue()

    def to_yaml_str(self):
        import yaml

        class ConfitYamlDumper(yaml.SafeDumper):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)

            def represent_ref(self, node):
                return super().represent_scalar("tag:yaml.org,2002:str", str(node))

            def represent_str(self, data):
                node = super().represent_scalar("tag:yaml.org,2002:str", data)
                if set(",'\"{}[]$") & set(data):
                    # node.value = dumps(data)
                    node.style = "'" if data.count('"') > data.count("'") else '"'
                return node

            yaml_representers = {
                **yaml.SafeDumper.yaml_representers,
                Config: yaml.SafeDumper.represent_dict,
                Reference: represent_ref,
                str: represent_str,
            }

        return yaml.dump(
            self.serialize(), Dumper=ConfitYamlDumper, sort_keys=False, indent=4
        )

    to_str = to_cfg_str

    def resolve(self, deep=True, registry: Any = None, root: Mapping = None) -> Any:
        """
        Resolves the parts of the nested config object with @ variables using
        a registry, and then interpolate references in the config.

        Parameters
        ----------
        deep: bool
            Should we resolve deeply
        registry:
            Registry to use when resolving
        root: Mapping
            The root of the config tree. Used for resolving references.

        Returns
        -------
        Union[Config, Any]
        """
        if root is None:
            root = self

        if registry is None:
            from .registry import get_default_registry

            registry = get_default_registry()
        resolved_locs = {}
        seen_locs = set()

        def resolve_reference(ref: Reference) -> Any:
            pat = re.compile(PATH + ":?")

            def replace(match: re.Match):
                start = match.start()
                if start > 0 and ref.value[start - 1] == ":":
                    return match.group()

                path = match.group()
                parts = split_path(path.rstrip(":"))
                try:
                    return local_names[parts] + ("." if path.endswith(":") else "")
                except KeyError:
                    raise KeyError(path)

            local_leaves = {}
            local_names = {}
            for match in pat.finditer(ref.value):
                start = match.start()
                if start > 0 and ref.value[start - 1] == ":":
                    continue
                path = match.group()
                parts = split_path(path.rstrip(":"))
                current = root
                for part in parts:
                    current = current[part]
                if id(current) not in resolved_locs:
                    resolved = rec(current, parts)
                else:
                    resolved = resolved_locs[id(current)]
                local_names[parts] = f"var_{len(local_leaves)}"
                local_leaves[f"var_{len(local_leaves)}"] = resolved

            replaced = pat.sub(replace, ref.value)

            res = safe_eval(replaced, local_leaves)

            return res

        def rec(obj, loc: Tuple[Union[str, int]] = ()):
            """
            Parameters
            ----------
            obj: Any
                The current object being resolved
            loc: Sequence[str]
                Internal variable
                Current path in tree

            Returns
            -------

            """
            if id(obj) in resolved_locs:
                return resolved_locs[id(obj)]

            if id(obj) in seen_locs:
                raise CyclicReferenceError(tuple(loc))

            seen_locs.add(id(obj))

            if not deep and len(loc) > 1:
                return obj

            if isinstance(obj, Mapping):
                resolved = Config({k: rec(v, (*loc, k)) for k, v in obj.items()})

                registries = [
                    (key, value, getattr(registry, key[1:]))
                    for key, value in resolved.items()
                    if key.startswith("@")
                ]
                assert (
                    len(registries) <= 1
                ), f"Cannot resolve using multiple registries at {'.'.join(loc)}"

                if len(registries) == 1:
                    cfg = resolved
                    params = dict(resolved)
                    params.pop(registries[0][0])
                    fn = registries[0][2].get(registries[0][1])
                    try:
                        resolved = fn(**params)
                        # The `validate_arguments` decorator has most likely
                        # already put the resolved config in the registry
                        # but for components that are instantiated without it
                        # we need to do it here
                        Config._store_resolved(resolved, cfg)
                    except ConfitValidationError as e:
                        e = ConfitValidationError(
                            errors=patch_errors(e.raw_errors, loc, params),
                            model=e.model,
                            name=getattr(e, "name", None),
                        ).with_traceback(remove_lib_from_traceback(e.__traceback__))
                        if not is_debug():
                            e.__cause__ = None
                            e.__suppress_context__ = True
                        raise e

            elif isinstance(obj, list):
                resolved = [rec(v, (*loc, i)) for i, v in enumerate(obj)]
            elif isinstance(obj, tuple):
                resolved = tuple(rec(v, (*loc, i)) for i, v in enumerate(obj))
            elif isinstance(obj, Reference):
                resolved = None
                while resolved is None:
                    try:
                        resolved = resolve_reference(obj)
                    except KeyError:
                        raise MissingReference(obj)
            else:
                resolved = obj

            resolved_locs[id(obj)] = resolved

            return resolved

        return rec(self, ())

    def merge(
        self,
        *updates: Union[Dict[str, Any], "Config"],
        remove_extra: bool = False,
    ) -> "Config":
        """
        Deep merge two configs. Heavily inspired from `thinc`'s config merge function.

        Parameters
        ----------
        updates: Union[Config, Dict]
            Configs to update the original config
        remove_extra:
            If true, restricts update to keys that existed in the original config

        Returns
        -------
        The new config
        """

        def deep_set(current, path, val):
            if path not in current and remove_extra:
                return
            current[path] = val

        def rec(old, new):
            for key, new_val in list(new.items()):
                if "." in key:
                    deep_set(old, key, new_val)
                    continue

                if key not in old:
                    if remove_extra:
                        continue
                    else:
                        old[key] = new_val
                        continue

                old_val = old[key]
                if isinstance(old_val, dict) and isinstance(new_val, dict):
                    old_resolver = next((k for k in old_val if k.startswith("@")), None)
                    new_resolver = next((k for k in new_val if k.startswith("@")), None)
                    if (
                        new_resolver is not None
                        and old_resolver is not None
                        and (
                            old_resolver != new_resolver
                            or old_val.get(old_resolver) != new_val.get(new_resolver)
                        )
                    ):
                        old[key] = new_val
                    else:
                        rec(old[key], new_val)
                else:
                    old[key] = new_val
            return old

        config = self.copy()
        for u in updates:
            rec(config, u)
        return config

    def copy(self: T) -> T:
        """
        Deep copy of the config, but not of the underlying data.
        Should also work with other types of objects (e.g. lists, tuples, etc.)

        ```
        Config.copy([1, 2, {"ok": 3}}]) == [1, 2, {"ok": 3}]
        ```

        Returns
        -------
        Any
        """
        seen = {}

        def rec(obj):
            if id(obj) in seen:
                return seen[id(obj)]
            seen[id(obj)] = obj
            if isinstance(obj, (Config, dict)):
                return type(obj)(
                    {k: rec(v) for k, v in obj.items()},
                )
            elif isinstance(obj, list):
                return [rec(v) for v in obj]
            elif isinstance(obj, tuple):
                return tuple(rec(v) for v in obj)
            elif isinstance(obj, Reference):
                return Reference(obj.value)
            else:
                return obj

        copy = rec(self)
        return copy

    @classmethod
    def _store_resolved(cls, resolved: Any, config: Dict[str, Any]):
        """
        Adds a resolved object to the RESOLVED_TO_CONFIG dict
        for later retrieval during serialization
        ([`.serialize`][confit.config.Config.serialize])

        Parameters
        ----------
        resolved: Any
        config: Config
        """
        try:
            RESOLVED_TO_CONFIG[resolved] = config
        except TypeError:
            pass

__init__(*args, **kwargs)

A new config object can be instantiated either from a dict as a positional argument, or from keyword arguments. Only one of these two options can be used at a time.

PARAMETER DESCRIPTION
args

TYPE: Any DEFAULT: ()

kwargs

TYPE: Any DEFAULT: {}

Source code in confit/config.py
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self, *args: Any, **kwargs: Any):
    """
    A new config object can be instantiated either from a dict as a positional
    argument, or from keyword arguments. Only one of these two options can be
    used at a time.

    Parameters
    ----------
    args: Any
    kwargs: Any
    """
    if len(args) == 1 and isinstance(args[0], dict):
        assert len(kwargs) == 0
        kwargs = args[0]
    super().__init__(**kwargs)

from_cfg_str(s, resolve=False, registry=None) classmethod

Load a config object from a config string

PARAMETER DESCRIPTION
s

The cfg config string

TYPE: str

resolve

Whether to resolve sections with '@' keys

TYPE: bool DEFAULT: False

registry

Optional registry to resolve from. If None, the default registry will be used.

TYPE: Any DEFAULT: None

RETURNS DESCRIPTION
Config
Source code in confit/config.py
 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
@classmethod
def from_cfg_str(cls, s: str, resolve: bool = False, registry: Any = None) -> Any:
    """
    Load a config object from a config string

    Parameters
    ----------
    s: Union[str, Path]
        The cfg config string
    resolve
        Whether to resolve sections with '@' keys
    registry
        Optional registry to resolve from.
        If None, the default registry will be used.

    Returns
    -------
    Config
    """
    parser = ConfigParser(interpolation=None)
    parser.optionxform = str
    parser.read_string(s)

    config = Config()

    for section in parser.sections():
        parts = split_path(section)
        current = config
        for part in parts:
            if part not in current:
                current[part] = current = Config()
            else:
                current = current[part]

        current.clear()
        errors = []
        for k, v in parser.items(section):
            path = split_path(k)
            for part in path[:-1]:
                if part not in current:
                    current[part] = current = Config()
                else:
                    current = current[part]
            try:
                current[path[-1]] = loads(v)
            except ValueError as e:
                errors.append(ErrorWrapper(e, loc=path))

        if errors:
            raise ConfitValidationError(errors=errors)

    if resolve:
        return config.resolve(registry=registry)

    return config

from_disk(path, resolve=False, registry=None) classmethod

Load a config object from a '.cfg' file

PARAMETER DESCRIPTION
path

The path to the config object

TYPE: Union[str, Path]

resolve

Whether to resolve mappings with '@' keys

TYPE: bool DEFAULT: False

registry

Optional registry to resolve from. If None, the default registry will be used.

TYPE: Any DEFAULT: None

RETURNS DESCRIPTION
Config
Source code in confit/config.py
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@classmethod
def from_disk(
    cls, path: Union[str, Path], resolve: bool = False, registry: Any = None
) -> "Config":
    """
    Load a config object from a '.cfg' file

    Parameters
    ----------
    path: Union[str, Path]
        The path to the config object
    resolve
        Whether to resolve mappings with '@' keys
    registry
        Optional registry to resolve from.
        If None, the default registry will be used.

    Returns
    -------
    Config
    """
    s = Path(path).read_text(encoding="utf-8")
    path_str = str(path)
    if path_str.endswith(".yaml") or path_str.endswith(".yml"):
        return cls.from_yaml_str(s, resolve=resolve, registry=registry)
    else:
        return cls.from_cfg_str(s, resolve=resolve, registry=registry)

to_disk(path)

Export a config to the disk (usually to a .cfg file)

PARAMETER DESCRIPTION
path

TYPE: Union[str, Path]

Source code in confit/config.py
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def to_disk(self, path: Union[str, Path]):
    """
    Export a config to the disk (usually to a .cfg file)

    Parameters
    ----------
    path: Union[str, path]
    """
    path_str = str(path)
    if path_str.endswith(".yaml") or path_str.endswith(".yml"):
        s = self.to_yaml_str()
    else:
        s = Config.to_str(self)
    Path(path).write_text(s)

serialize()

Try to convert non-serializable objects using the RESOLVED_TO_CONFIG object back to their original catalogue + params form

We try to preserve referential equalities between non dict/list/tuple objects by serializing subsequent references to the same object as references to its first occurrence in the tree.

a = A()  # serializable object
cfg = {"a": a, "b": a}
print(Config.serialize(cfg))
# Out: {"a": {...}, "b": Reference("a")}
RETURNS DESCRIPTION
Config
Source code in confit/config.py
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def serialize(self: Any):
    """
    Try to convert non-serializable objects using the RESOLVED_TO_CONFIG object
    back to their original catalogue + params form

    We try to preserve referential equalities between non dict/list/tuple
    objects by serializing subsequent references to the same object as references
    to its first occurrence in the tree.

    ```python
    a = A()  # serializable object
    cfg = {"a": a, "b": a}
    print(Config.serialize(cfg))
    # Out: {"a": {...}, "b": Reference("a")}
    ```

    Returns
    -------
    Config
    """
    refs = {}

    # Temp memory to avoid objects being garbage collected
    mem = []

    def is_simple(o):
        return o is None or isinstance(o, (str, int, float, bool, Reference))

    def rec(o: Any, path: Loc = ()):
        if id(o) in refs:
            return refs[id(o)]
        if is_simple(o):
            return o
        if isinstance(o, collections.abc.Mapping):
            items = sorted(
                o.items(),
                key=lambda x: 1
                if (
                    is_simple(x[1])
                    or isinstance(x[1], (collections.abc.Mapping, list, tuple))
                )
                else 0,
            )
            serialized = {k: rec(v, (*path, k)) for k, v in items}
            serialized = {k: serialized[k] for k in o.keys()}
            mem.append(o)
            refs[id(o)] = Reference(join_path(path))
            if isinstance(o, Config):
                serialized = Config(serialized)
            return serialized
        if isinstance(o, (list, tuple)):
            mem.append(o)
            refs[id(o)] = Reference(join_path(path))
            return type(o)(rec(v, (*path, i)) for i, v in enumerate(o))
        cfg = None
        try:
            cfg = (cfg or Config()).merge(RESOLVED_TO_CONFIG[o])
        except (KeyError, TypeError):
            pass
        try:
            cfg = (cfg or Config()).merge(o.cfg)
        except (AttributeError, TypeError):
            pass
        if cfg is not None:
            mem.append(o)
            refs[id(o)] = Reference(join_path(path))
            return rec(cfg, path)
        try:
            return pydantic_core.to_jsonable_python(o)
        except Exception:
            raise TypeError(f"Cannot dump {o!r} at {join_path(path)}")

    return rec(self)

to_cfg_str()

Export a config to a string in the cfg format by serializing it first

RETURNS DESCRIPTION
str
Source code in confit/config.py
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def to_cfg_str(self):
    """
    Export a config to a string in the cfg format
    by serializing it first

    Returns
    -------
    str
    """
    additional_sections = {}

    prepared = flatten_sections(Config.serialize(self))
    prepared.update(flatten_sections(additional_sections))

    parser = ConfigParser(interpolation=None)
    parser.optionxform = str
    for section_name, section in prepared.items():
        parser.add_section(section_name)
        parser[section_name].update(
            {join_path((k,)): dumps(v) for k, v in section.items()}
        )
    s = StringIO()
    parser.write(s)
    return s.getvalue()

resolve(deep=True, registry=None, root=None)

Resolves the parts of the nested config object with @ variables using a registry, and then interpolate references in the config.

PARAMETER DESCRIPTION
deep

Should we resolve deeply

DEFAULT: True

registry

Registry to use when resolving

TYPE: Any DEFAULT: None

root

The root of the config tree. Used for resolving references.

TYPE: Mapping DEFAULT: None

RETURNS DESCRIPTION
Union[Config, Any]
Source code in confit/config.py
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
def resolve(self, deep=True, registry: Any = None, root: Mapping = None) -> Any:
    """
    Resolves the parts of the nested config object with @ variables using
    a registry, and then interpolate references in the config.

    Parameters
    ----------
    deep: bool
        Should we resolve deeply
    registry:
        Registry to use when resolving
    root: Mapping
        The root of the config tree. Used for resolving references.

    Returns
    -------
    Union[Config, Any]
    """
    if root is None:
        root = self

    if registry is None:
        from .registry import get_default_registry

        registry = get_default_registry()
    resolved_locs = {}
    seen_locs = set()

    def resolve_reference(ref: Reference) -> Any:
        pat = re.compile(PATH + ":?")

        def replace(match: re.Match):
            start = match.start()
            if start > 0 and ref.value[start - 1] == ":":
                return match.group()

            path = match.group()
            parts = split_path(path.rstrip(":"))
            try:
                return local_names[parts] + ("." if path.endswith(":") else "")
            except KeyError:
                raise KeyError(path)

        local_leaves = {}
        local_names = {}
        for match in pat.finditer(ref.value):
            start = match.start()
            if start > 0 and ref.value[start - 1] == ":":
                continue
            path = match.group()
            parts = split_path(path.rstrip(":"))
            current = root
            for part in parts:
                current = current[part]
            if id(current) not in resolved_locs:
                resolved = rec(current, parts)
            else:
                resolved = resolved_locs[id(current)]
            local_names[parts] = f"var_{len(local_leaves)}"
            local_leaves[f"var_{len(local_leaves)}"] = resolved

        replaced = pat.sub(replace, ref.value)

        res = safe_eval(replaced, local_leaves)

        return res

    def rec(obj, loc: Tuple[Union[str, int]] = ()):
        """
        Parameters
        ----------
        obj: Any
            The current object being resolved
        loc: Sequence[str]
            Internal variable
            Current path in tree

        Returns
        -------

        """
        if id(obj) in resolved_locs:
            return resolved_locs[id(obj)]

        if id(obj) in seen_locs:
            raise CyclicReferenceError(tuple(loc))

        seen_locs.add(id(obj))

        if not deep and len(loc) > 1:
            return obj

        if isinstance(obj, Mapping):
            resolved = Config({k: rec(v, (*loc, k)) for k, v in obj.items()})

            registries = [
                (key, value, getattr(registry, key[1:]))
                for key, value in resolved.items()
                if key.startswith("@")
            ]
            assert (
                len(registries) <= 1
            ), f"Cannot resolve using multiple registries at {'.'.join(loc)}"

            if len(registries) == 1:
                cfg = resolved
                params = dict(resolved)
                params.pop(registries[0][0])
                fn = registries[0][2].get(registries[0][1])
                try:
                    resolved = fn(**params)
                    # The `validate_arguments` decorator has most likely
                    # already put the resolved config in the registry
                    # but for components that are instantiated without it
                    # we need to do it here
                    Config._store_resolved(resolved, cfg)
                except ConfitValidationError as e:
                    e = ConfitValidationError(
                        errors=patch_errors(e.raw_errors, loc, params),
                        model=e.model,
                        name=getattr(e, "name", None),
                    ).with_traceback(remove_lib_from_traceback(e.__traceback__))
                    if not is_debug():
                        e.__cause__ = None
                        e.__suppress_context__ = True
                    raise e

        elif isinstance(obj, list):
            resolved = [rec(v, (*loc, i)) for i, v in enumerate(obj)]
        elif isinstance(obj, tuple):
            resolved = tuple(rec(v, (*loc, i)) for i, v in enumerate(obj))
        elif isinstance(obj, Reference):
            resolved = None
            while resolved is None:
                try:
                    resolved = resolve_reference(obj)
                except KeyError:
                    raise MissingReference(obj)
        else:
            resolved = obj

        resolved_locs[id(obj)] = resolved

        return resolved

    return rec(self, ())

merge(*updates, remove_extra=False)

Deep merge two configs. Heavily inspired from thinc's config merge function.

PARAMETER DESCRIPTION
updates

Configs to update the original config

TYPE: Union[Dict[str, Any], Config] DEFAULT: ()

remove_extra

If true, restricts update to keys that existed in the original config

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
The new config
Source code in confit/config.py
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
def merge(
    self,
    *updates: Union[Dict[str, Any], "Config"],
    remove_extra: bool = False,
) -> "Config":
    """
    Deep merge two configs. Heavily inspired from `thinc`'s config merge function.

    Parameters
    ----------
    updates: Union[Config, Dict]
        Configs to update the original config
    remove_extra:
        If true, restricts update to keys that existed in the original config

    Returns
    -------
    The new config
    """

    def deep_set(current, path, val):
        if path not in current and remove_extra:
            return
        current[path] = val

    def rec(old, new):
        for key, new_val in list(new.items()):
            if "." in key:
                deep_set(old, key, new_val)
                continue

            if key not in old:
                if remove_extra:
                    continue
                else:
                    old[key] = new_val
                    continue

            old_val = old[key]
            if isinstance(old_val, dict) and isinstance(new_val, dict):
                old_resolver = next((k for k in old_val if k.startswith("@")), None)
                new_resolver = next((k for k in new_val if k.startswith("@")), None)
                if (
                    new_resolver is not None
                    and old_resolver is not None
                    and (
                        old_resolver != new_resolver
                        or old_val.get(old_resolver) != new_val.get(new_resolver)
                    )
                ):
                    old[key] = new_val
                else:
                    rec(old[key], new_val)
            else:
                old[key] = new_val
        return old

    config = self.copy()
    for u in updates:
        rec(config, u)
    return config

copy()

Deep copy of the config, but not of the underlying data. Should also work with other types of objects (e.g. lists, tuples, etc.)

Config.copy([1, 2, {"ok": 3}}]) == [1, 2, {"ok": 3}]
RETURNS DESCRIPTION
Any
Source code in confit/config.py
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
def copy(self: T) -> T:
    """
    Deep copy of the config, but not of the underlying data.
    Should also work with other types of objects (e.g. lists, tuples, etc.)

    ```
    Config.copy([1, 2, {"ok": 3}}]) == [1, 2, {"ok": 3}]
    ```

    Returns
    -------
    Any
    """
    seen = {}

    def rec(obj):
        if id(obj) in seen:
            return seen[id(obj)]
        seen[id(obj)] = obj
        if isinstance(obj, (Config, dict)):
            return type(obj)(
                {k: rec(v) for k, v in obj.items()},
            )
        elif isinstance(obj, list):
            return [rec(v) for v in obj]
        elif isinstance(obj, tuple):
            return tuple(rec(v) for v in obj)
        elif isinstance(obj, Reference):
            return Reference(obj.value)
        else:
            return obj

    copy = rec(self)
    return copy

merge_from_disk(config_paths, returned_name='first')

Merge multiple configs loaded from the filesystem and return the merged config as well as the name of the config

PARAMETER DESCRIPTION
config_paths

Paths to the config files

TYPE: Union[Path, List[Path]]

returned_name

If "first", the name of the first config is returned as the name of the merged config. If "concat", the names of the configs are concatenated with a "+" sign

TYPE: str DEFAULT: 'first'

Source code in confit/config.py
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
def merge_from_disk(
    config_paths: Union[Path, List[Path]],
    returned_name: str = "first",
):
    """
    Merge multiple configs loaded from the filesystem
    and return the merged config as well as the name of the config

    Parameters
    ----------
    config_paths: Union[Path, List[Path]]
        Paths to the config files
    returned_name: str
        If "first", the name of the first config is returned as the name of the merged
        config. If "concat", the names of the configs are concatenated with a "+" sign

    Returns
    -------

    """
    assert returned_name in {"first", "concat"}
    if isinstance(config_paths, Path):
        config_paths = [config_paths]

    configs = [Config.from_disk(p, resolve=False) for p in config_paths]
    config_names = [p.stem for p in config_paths]

    name = config_names[0] if returned_name == "first" else "+".join(config_names)

    config = configs.pop(0)
    return config.merge(*configs), name