Source code for jgdv.structs.chainguard.chainguard

  1#/usr/bin/env python3
  2"""
  3
  4"""
  5# Imports:
  6from __future__ import annotations
  7
  8# ##-- stdlib imports
  9import datetime
 10import enum
 11import functools as ftz
 12import itertools as itz
 13import logging as logmod
 14import pathlib as pl
 15from collections import ChainMap
 16from copy import deepcopy
 17from dataclasses import InitVar, dataclass, field
 18from re import Pattern
 19from uuid import UUID, uuid1
 20from weakref import ref
 21import tomllib
 22# ##-- end stdlib imports
 23
 24# ##-- 1st party imports
 25from jgdv import Proto, Mixin
 26
 27from ._base import GuardBase
 28from .errors import GuardedAccessError
 29from .mixins.access_m import TomlAccess_m
 30from .mixins.loader_m import TomlLoader_m
 31from .mixins.proxy_m import GuardProxyEntry_m
 32from .mixins.reporter_m import DefaultedReporter_m
 33from .mixins.writer_m import TomlWriter_m
 34
 35from ._interface import TomlTypes, ChainGuard_p
 36# ##-- end 1st party imports
 37
 38# ##-- types
 39# isort: off
 40import abc
 41import collections.abc
 42from typing import TYPE_CHECKING, cast, assert_type, assert_never
 43from typing import Generic, NewType
 44# Protocols:
 45from typing import Protocol, runtime_checkable
 46# Typing Decorators:
 47from typing import no_type_check, final, override, overload
 48
 49if TYPE_CHECKING:
 50    from jgdv import Maybe
 51    from typing import Final
 52    from typing import ClassVar, Any, LiteralString
 53    from typing import Never, Self, Literal
 54    from typing import TypeGuard
 55    from collections.abc import Iterable, Iterator, Callable, Generator
 56    from collections.abc import Sequence, Mapping, MutableMapping, Hashable
 57
 58##--|
 59
 60# isort: on
 61# ##-- end types
 62
 63##-- logging
 64logging = logmod.getLogger(__name__)
 65##-- end logging
 66
 67##--|
 68
[docs] 69@Proto(ChainGuard_p) 70@Mixin(TomlLoader_m, TomlWriter_m, silent=True) 71@Mixin(DefaultedReporter_m, silent=True) 72class ChainGuard(GuardProxyEntry_m, GuardBase): 73 """ The Final ChainGuard class. 74 75 Takes the GuardBase object, and mixes in extra capabilities. 76 77 """
[docs] 78 @classmethod 79 def merge(cls, *guards:Self, dfs:Maybe[Callable]=None, index:Maybe[str]=None, shadow:bool=False) -> Self: # noqa: ARG003 80 """ 81 Given an ordered list of guards and dicts, convert them to dicts, 82 update an empty dict with each, 83 then wrap that combined dict in a ChainGuard 84 85 *NOTE*: classmethod, not instance. search order is same as arg order. 86 So merge(a, b, c) will retrive from c only if a, then b, don't have the key 87 88 # TODO if given a dfs callable, use it to merge more intelligently 89 """ 90 curr_keys : set = set() 91 # Check for conflicts: 92 for data in guards: 93 new_keys = set(data.keys()) 94 if bool(curr_keys & new_keys) and not shadow: 95 msg = "Key Conflict:" 96 raise KeyError(msg, curr_keys & new_keys) 97 curr_keys |= new_keys 98 99 # Build a TG from a chainmap 100 return ChainGuard.from_dict(ChainMap(*(dict(x) for x in guards))) # type: ignore[attr-defined]
101
[docs] 102 def remove_prefix(self, prefix:str) -> ChainGuard: 103 """ Try to remove a prefix from loaded data 104 eg: ChainGuard(tools.ChainGuard.data..).remove_prefix("tools.ChainGuard") 105 -> ChainGuard(data..) 106 """ 107 match prefix: 108 case None: 109 return self 110 case str(): 111 logging.debug("Removing prefix from data: %s", prefix) 112 try: 113 attempt = self 114 for x in prefix.split("."): 115 attempt = attempt[x] 116 else: 117 return ChainGuard(attempt) 118 except GuardedAccessError: 119 return self
120 121