Source code for jgdv.structs.chainguard.proxies.base

  1#!/usr/bin/env python3
  2"""
  3
  4"""
  5
  6# Imports:
  7from __future__ import annotations
  8
  9# ##-- stdlib imports
 10import contextlib
 11import datetime
 12import enum
 13import faulthandler
 14import functools as ftz
 15import hashlib
 16import itertools as itz
 17import logging as logmod
 18import pathlib as pl
 19import re
 20import time
 21import types
 22import weakref
 23from copy import deepcopy
 24from time import sleep
 25from uuid import UUID, uuid1
 26from weakref import ref
 27
 28# ##-- end stdlib imports
 29
 30# ##-- 1st party imports
 31from .._base import GuardBase
 32from jgdv.structs.chainguard.mixins.reporter_m import DefaultedReporter_m
 33
 34# ##-- end 1st party imports
 35
 36# ##-- types
 37# isort: off
 38import abc
 39import collections.abc
 40from typing import TYPE_CHECKING, cast, assert_type, assert_never
 41from typing import Generic, NewType
 42# Protocols:
 43from typing import Protocol, runtime_checkable
 44# Typing Decorators:
 45from typing import no_type_check, final, override, overload
 46from typing import Any, Never
 47
 48if TYPE_CHECKING:
 49    from .._interface import TomlTypes
 50    from jgdv import Maybe
 51    from typing import Final
 52    from typing import ClassVar, LiteralString
 53    from typing import 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    from jgdv import CHECKTYPE
 59    from .._interface import ChainGuard_i, ProxyWrapper
 60
 61# isort: on
 62# ##-- end types
 63
 64##-- logging
 65logging = logmod.getLogger(__name__)
 66##-- end logging
 67
 68ROOT_INDEX : Final[tuple[str, ...]] = ("<root>", )
 69##--|
[docs] 70class GuardProxy: 71 """ A Base Class for Proxies """ 72 __slots__ = ("__index", "_data", "_fallback", "_types") 73 __index : tuple[str|int, ...] 74 _data : Maybe[TomlTypes|GuardBase] 75 _types : CHECKTYPE 76 _fallback : Maybe[Any|tuple] 77 78 def __init__(self, data:Maybe[GuardBase|Any], types:Any=None, index:Maybe[Iterable[str|int]]=None, fallback:Maybe[tuple|Any]=()) -> None: # noqa: ANN401 79 self._data = data 80 self._types = types 81 self._fallback = fallback 82 self.__index = tuple(index or ROOT_INDEX) 83 84 @override 85 def __repr__(self) -> str: 86 type_str = self._types_str() 87 index_str = ".".join(map(str, self._index())) 88 return f"<GuardProxy: {index_str}:{type_str}>" 89 90 def __len__(self) -> int: 91 if isinstance(self._data, collections.abc.Sized): 92 return len(self._data) 93 94 return 0 95 96 def __bool__(self) -> bool: 97 return self._data is not None and self._data is not Never 98 99 def __call__(self, *, wrapper:Maybe[ProxyWrapper]=None, **kwargs:Any) -> Any: # noqa: ANN401 100 raise NotImplementedError() 101 102 def __getattr__(self, attr:str) -> GuardProxy: 103 raise NotImplementedError() 104 105 def __getitem__(self, keys:int|str|tuple[str]) -> GuardProxy: 106 raise NotImplementedError() 107 108 def __contains__(self, other): 109 match other: 110 case str(): 111 return other in self._data or other.replace("_","-") 112 case x: 113 return x in self._data 114 ##--| 115
[docs] 116 def _inject(self, val:Maybe=None, attr:Maybe[int|str|tuple[int|str, ...]]=None, *, clear:bool=False) -> GuardProxy: 117 """ create a new proxy that is further into the proxied data """ 118 match val: 119 case _ if clear: 120 val = None 121 case _: 122 pass 123 124 return type(self)(val, 125 types=self._types, 126 index=self._index(attr), 127 fallback=self._fallback)
128
[docs] 129 def _notify(self) -> None: 130 types_str = self._types_str() 131 match self._data, self._fallback, self._index(): 132 case GuardBase(), _, _: 133 pass 134 case None, (), _: 135 pass 136 case _, _, []: 137 pass 138 case None, val, [*index]: 139 DefaultedReporter_m.add_defaulted(".".join(map(str, index)), 140 val, 141 types_str) 142 case val, _, [*index]: 143 assert(not isinstance(val, GuardBase|None)) 144 DefaultedReporter_m.add_defaulted(".".join(map(str, index)), 145 val, 146 types_str) 147 case val, flbck, index,: 148 msg = "Unexpected Values found: " 149 raise TypeError(msg, val, index, flbck)
150
[docs] 151 def _types_str(self) -> str: 152 types_str : str 153 match self._types: 154 case None: 155 return "Any" 156 case types.UnionType() as targ: 157 types_str = repr(targ) 158 case type(__name__=targ): 159 types_str = targ 160 case _ as targ: 161 types_str = str(targ) 162 163 return types_str
164
[docs] 165 def _match_type(self, val:Maybe) -> Maybe: 166 match self._types: 167 case type() if not isinstance(val, self._types): 168 types_str = self._types_str() 169 index_str = ".".join(*map(str, self._index(f"({types_str})"))) 170 msg = "TomlProxy Value doesn't match declared Type: " 171 raise TypeError(msg, index_str, val, self._types) 172 case _: 173 # TODO handle unions and generics 174 pass 175 176 return val
177
[docs] 178 def _index(self, sub:Maybe[int|str|tuple[int|str, ...]]=None) -> tuple[str|int, ...]: 179 """ create a new index from this proxy """ 180 match sub: 181 case None: 182 return self.__index 183 case int()|str() as x: 184 return (*self.__index, x) 185 case [*xs]: 186 return (*self.__index, *xs)