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

  1#!/usr/bin/env python3
  2"""
  3A Proxy for ChainGuard,
  4  which allows you to use the default attribute access
  5  (data.a.b.c)
  6  even when there might not be an `a.b.c` path in the data.
  7
  8  Thus:
  9  data.on_fail(default_value).a.b.c()
 10
 11  Note: To distinguish between not giving a default value,
 12  and giving a default value of `None`,
 13  wrap the default value in a tuple: (None,)
 14"""
 15
 16# Imports:
 17from __future__ import annotations
 18
 19# ##-- stdlib imports
 20import atexit#  for @atexit.register
 21import collections
 22import contextlib
 23import datetime
 24import enum
 25import faulthandler
 26import functools as ftz
 27import hashlib
 28import itertools as itz
 29import logging as logmod
 30import pathlib as pl
 31import re
 32import time
 33import types as types_
 34import weakref
 35from copy import deepcopy
 36from time import sleep
 37from uuid import UUID, uuid1
 38from weakref import ref
 39
 40# ##-- end stdlib imports
 41
 42# ##-- 1st party imports
 43from jgdv import Proto
 44from .._base import GuardBase
 45from .._interface import TomlTypes, ChainProxy_p
 46from ..errors import GuardedAccessError
 47from .base import GuardProxy
 48
 49# ##-- end 1st party imports
 50
 51# ##-- types
 52# isort: off
 53import abc
 54import collections.abc
 55from typing import TYPE_CHECKING, cast, assert_type, assert_never
 56from typing import Generic, NewType
 57# Protocols:
 58from typing import Protocol, runtime_checkable
 59# Typing Decorators:
 60from typing import no_type_check, final, override, overload
 61from typing import Never
 62
 63if TYPE_CHECKING:
 64    from jgdv import Maybe
 65    from typing import Final
 66    from typing import ClassVar, Any, LiteralString
 67    from typing import Self, Literal
 68    from typing import TypeGuard
 69    from collections.abc import Iterable, Iterator, Callable, Generator
 70    from collections.abc import Sequence, Mapping, MutableMapping, Hashable
 71
 72    from .._interface import ChainGuard_i
 73
 74    type Wrapper = Callable[[TomlTypes], Any]
 75
 76# isort: on
 77# ##-- end types
 78
 79##-- logging
 80logging = logmod.getLogger(__name__)
 81##-- end logging
 82
 83NO_FALLBACK : Final[tuple] = ()
 84##--|
 85
[docs] 86@Proto(ChainProxy_p) 87class GuardFailureProxy(GuardProxy): 88 """ 89 A Wrapper for guarded access to toml values. 90 you get the value by calling it. 91 Until then, it tracks attribute access, 92 and reports that to GuardBase when called. 93 It also can type check its value and the value retrieved from the toml data 94 """ 95 96 def __init__(self, data:Maybe, types:Maybe=None, index:Maybe[list[str]]=None, fallback:Maybe[TomlTypes|tuple]=NO_FALLBACK) -> None: 97 super().__init__(data, types=types, index=index, fallback=fallback) 98 match self._fallback: 99 case tuple(): 100 pass 101 case _: 102 self._match_type(self._fallback) 103 104 @override 105 def __call__(self, wrapper:Maybe[Wrapper]=None, fallback_wrapper:Maybe[Wrapper]=None, **kwargs:Any) -> Any: 106 """ 107 Reify a proxy into an actual value, or its fallback. 108 Optionally call a wrapper function on the actual value, 109 or a fallback_wrapper function on the fallback 110 """ 111 val : Any 112 ##--| 113 self._notify() 114 wrapper = wrapper or (lambda x: x) 115 fallback_wrapper = fallback_wrapper or (lambda x: x) 116 match self._data, self._fallback: 117 case None, tuple() as x if x == (): 118 msg = "No Value, and no fallback" 119 raise ValueError(msg) 120 case GuardBase() as data, _: 121 val = wrapper(data) 122 case None, data: 123 val = fallback_wrapper(data) # type: ignore[arg-type] 124 case _ as data, _: 125 val = wrapper(data) # type: ignore[arg-type] 126 127 return self._match_type(val) 128 129 @override 130 def __getattr__(self, attr:str) -> GuardProxy: 131 return self.__getitem__(attr) 132 133 @override 134 def __getitem__(self, keys:int|str|tuple[int|str, ...]) -> GuardProxy: 135 curr : GuardProxy = self 136 match keys: 137 case tuple(): 138 pass 139 case str() | int(): 140 keys = (keys,) 141 case x: 142 raise TypeError(type(x)) 143 144 curr = self 145 try: 146 for x in keys: 147 match curr._data, x: 148 case None, _: 149 raise GuardedAccessError() # noqa: TRY301 150 case dict() as d, k if k in d: 151 curr = curr._inject(d[k], attr=x) 152 case list() as d, int() as k if k < len(d): 153 curr = curr._inject(d[k], attr=k) 154 case _: 155 curr = curr._inject(attr=x) 156 except GuardedAccessError: 157 return curr._inject(clear=True, attr=keys) 158 else: 159 return curr