TracingProxy

tracing_proxy.py

from __future__ import annotations

from collections.abc import Iterable, Mapping, Sequence
from numbers import Number
from typing import Any


def _should_wrap(value: Any) -> bool:
    return not isinstance(value, (str, bytes, Number, bool)) and value is not None


class TracingProxy(dict):
    def __init__(
        self,
        *,
        value: Any,
        path: str = "",
        accesses: set[str] | None = None,
    ) -> None:
        super().__init__(value) if isinstance(value, Mapping) else super().__init__()
        self._value = value
        self._path = path
        self._accesses = accesses if accesses is not None else set()

    def _child_path(self, key: Any) -> str:
        segment = f"[{key}]" if isinstance(key, int) else str(key)
        if not self._path:
            return segment
        return f"{self._path}{segment}" if segment.startswith("[") else f"{self._path}.{segment}"

    def _record(self, path: str) -> None:
        if path:
            self._accesses.add(path)

    def _wrap(self, path: str, value: Any) -> Any:
        if callable(value):
            return value
        return (
            TracingProxy(value=value, path=path, accesses=self._accesses)
            if _should_wrap(value)
            else value
        )

    def __getitem__(self, key: Any) -> Any:  # type: ignore[override]
        path = self._child_path(key)
        self._record(path)
        value = self._value
        try:
            if isinstance(value, (Mapping, Sequence)) and not isinstance(value, (str, bytes)):
                child = value[key]
            else:
                child = value[key]  # type: ignore[index]
        except Exception:
            raise KeyError(key)
        return self._wrap(path, child)

    def __getattr__(self, key: str) -> Any:
        if key.startswith("_"):
            return super().__getattribute__(key)
        path = self._child_path(key)
        self._record(path)
        value = self._value
        try:
            if isinstance(value, Mapping) and key in value:
                child = value[key]
            else:
                child = getattr(value, key)
        except AttributeError:
            raise AttributeError(key)
        return child if callable(child) else self._wrap(path, child)

    def get(self, key: Any, default: Any = None) -> Any:
        try:
            return self[key]
        except KeyError:
            return default

    def __contains__(self, key: Any) -> bool:  # type: ignore[override]
        try:
            return key in self._value  # type: ignore[operator]
        except TypeError:
            return False

    def __iter__(self):  # type: ignore[override]
        value = self._value
        if isinstance(value, Mapping):
            return iter(value)
        if isinstance(value, Iterable) and not isinstance(value, (str, bytes)):

            def generator():
                for idx, item in enumerate(value):
                    path = self._child_path(idx)
                    self._record(path)
                    yield self._wrap(path, item)

            return generator()
        raise TypeError(f"Object of type {type(value)!r} is not iterable")

    def __len__(self) -> int:  # type: ignore[override]
        try:
            return len(self._value)  # type: ignore[arg-type]
        except TypeError:
            return 0

    def __bool__(self) -> bool:  # type: ignore[override]
        return bool(self._value)

    def __str__(self) -> str:  # type: ignore[override]
        return str(self._value)

    def __repr__(self) -> str:  # type: ignore[override]
        return f"TracingProxy({self._value!r})"

    def __eq__(self, other: object) -> bool:  # type: ignore[override]
        return self._value == (other._value if isinstance(other, TracingProxy) else other)

    def __hash__(self) -> int:  # type: ignore[override]
        return hash(self._value)