update mirror of ShineyDev/utility (#6)

ShineyDev/utility@b245f4c

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

pretty/utility/_mirror/__init__.py

@@ -0,0 +1,38 @@
+"""
+utility: A Python package with utilities I use in several of my other projects.
+"""
+
+from __future__ import annotations
+from typing import NamedTuple
+
+from .cache import *
+from .cache import __all__ as _cache__all__
+from .typing import *
+from .typing import __all__ as _typing__all__
+from .version import *
+from .version import __all__ as _version__all__
+from .warning import *
+from .warning import __all__ as _warning__all__
+from .wrapper import *
+from .wrapper import __all__ as _wrapper__all__
+
+
+class _VersionInfo(NamedTuple):
+    major: int
+    minor: int
+    micro: int
+    release: str
+    serial: int
+
+
+version: str = "0.1.0a"
+version_info: _VersionInfo = _VersionInfo(0, 1, 0, "alpha", 0)
+
+
+__all__ = [  # pyright: ignore[reportUnsupportedDunderAll]
+    *_cache__all__,
+    *_typing__all__,
+    *_version__all__,
+    *_warning__all__,
+    *_wrapper__all__,
+]

pretty/utility/_mirror/cache.py

@@ -0,0 +1,186 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING, TypeVar
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Collection, Generator
+    from typing import Any, overload
+    from typing_extensions import TypeAlias, ParamSpec, Self
+
+    _P = ParamSpec("_P")
+    _T = TypeVar("_T")
+    _U = TypeVar("_U")
+
+    _Generator: TypeAlias = Generator[_T, None, Any]
+    _GeneratorFunc: TypeAlias = Callable[_P, Generator[_T, None, Any]]
+
+import collections
+import typing
+
+from .typing import MISSING
+from .version import SUPPORTS_GENERICBUILTINS
+
+
+def _make_key(
+    args: tuple[Any, ...],
+    kwargs: dict[str, Any],
+) -> int:
+    return hash((args, frozenset(kwargs.items())))
+
+
+if TYPE_CHECKING:
+
+    @overload
+    def cache_generator(
+        wrapped: _GeneratorFunc[_P, _T],
+        /,
+    ) -> _GeneratorFunc[_P, _T]: ...
+
+    @overload
+    def cache_generator(
+        *,
+        max_size: int | None = ...,
+    ) -> Callable[[_GeneratorFunc[_P, _T]], _GeneratorFunc[_P, _T]]: ...
+
+    @overload
+    def cache_generator(
+        *,
+        max_size: int | None = ...,
+        wrapper: Callable[[_Generator[_T]], _U],
+    ) -> Callable[[_GeneratorFunc[_P, _T]], Callable[_P, _U]]: ...
+
+
+def cache_generator(
+    wrapped: _GeneratorFunc[_P, _T] = MISSING,
+    /,
+    *,
+    max_size: int | None = MISSING,
+    wrapper: Callable[[_Generator[_T]], _U] = MISSING,
+) -> _GeneratorFunc[_P, _T] | Callable[[_GeneratorFunc[_P, _T]], _GeneratorFunc[_P, _T]] | Callable[[_GeneratorFunc[_P, _T]], Callable[_P, _U]]:
+    max_size = max_size if max_size is not MISSING else 1024
+
+    if isinstance(max_size, int):
+        if max_size < -1 or max_size == 0:
+            raise ValueError("max_size must be None, -1, or a positive integer")
+
+    if wrapper is not MISSING:
+
+        def decorator_wrapper(
+            wrapped: _GeneratorFunc[_P, _T],
+            /,
+        ) -> Callable[_P, _U]:
+            cache: dict[int, _U] | None
+
+            if max_size == -1 or max_size is None:
+                cache = dict()
+            else:
+                cache = LRUCache(max_size=max_size)
+
+            def inner(
+                *args: _P.args,
+                **kwargs: _P.kwargs,
+            ) -> _U:
+                key = _make_key(args, kwargs)
+
+                if key not in cache:
+                    cache[key] = wrapper(wrapped(*args, **kwargs))
+
+                return cache[key]
+
+            inner.__utility_cache__ = cache
+
+            return inner
+
+        return decorator_wrapper
+
+    else:
+
+        def decorator(
+            wrapped: _GeneratorFunc[_P, _T],
+            /,
+        ) -> _GeneratorFunc[_P, _T]:
+            cache: dict[int, tuple[Generator[_T, None, Any], list[_T], bool]]
+
+            if max_size == -1 or max_size is None:
+                cache = dict()
+            else:
+                cache = LRUCache(max_size=max_size)
+
+            def inner(
+                *args: _P.args,
+                **kwargs: _P.kwargs,
+            ) -> Generator[_T, None, Any]:
+                key = _make_key(args, kwargs)
+
+                if key not in cache:
+                    generator = wrapped(*args, **kwargs)
+                    cache[key] = (generator, list(), False)
+
+                generator, items, done = cache[key]
+
+                i = 0  # NOTE: this garbage is all required to support multiple entries before exit
+                while i < len(items):
+                    yield items[i]
+                    i += 1
+
+                if not done:
+                    i = 0
+                    for item in generator:
+                        items.append(item)
+                        yield item
+                        i += 1
+
+                        if cache[key][2]:
+                            yield from items[i:]
+                            return
+
+                    cache[key] = (MISSING, items, True)
+
+            inner.__utility_cache__ = cache
+
+            return inner
+
+        if wrapped is MISSING:
+            return decorator
+
+        return decorator(wrapped)
+
+
+_K = TypeVar("_K")
+_V = TypeVar("_V")
+
+
+class LRUCache(collections.OrderedDict[_K, _V] if SUPPORTS_GENERICBUILTINS else typing.OrderedDict[_K, _V]):  # type: ignore
+    """
+    TODO
+    """
+
+    def __init__(
+        self: Self,
+        /,
+        *,
+        max_size: int,
+    ) -> None:
+        super().__init__()
+
+        self.max_size = max_size
+
+    def __getitem__(self, key: _K) -> _V:
+        value = super().__getitem__(key)
+        self.move_to_end(key)
+
+        return value
+
+    def __setitem__(self, key: _K, value: _V) -> None:
+        if key in self:
+            self.move_to_end(key)
+
+        super().__setitem__(key, value)
+
+        if len(self) > self.max_size:
+            self.popitem(last=False)
+
+
+__all__ = [
+    "cache_generator",
+    "LRUCache",
+]

pretty/utility/_mirror/typing.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Any
+
+
+class _MissingSentinel:
+    __slots__ = ()
+
+    def __repr__(self) -> str:
+        return "..."
+
+    def __bool__(self) -> bool:
+        return False
+
+
+MISSING: Any = _MissingSentinel()
+"""
+TODO
+"""
+
+
+__all__ = [
+    "MISSING",
+]

pretty/utility/_mirror/version.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Final
+
+import importlib.util
+import sys
+
+
+try:
+    from typing_extensions import __all__ as _typing_extensions__all__
+except ImportError:
+    _typing_extensions__all__ = ()
+
+
+PY_39: Final[bool] = sys.version_info >= (3, 9, 0)
+
+SUPPORTS_ANNOTATED = PY_39 or ("Annotated" in _typing_extensions__all__)  # typing.Annotated
+SUPPORTS_GENERICBUILTINS = PY_39  # list[int], collections.abc.Sequence[int]
+SUPPORTS_ZONEINFO = PY_39 or (importlib.util.find_spec("backports") and importlib.util.find_spec("backports.zoneinfo")) is not None  # zoneinfo
+
+PY_310: Final[bool] = sys.version_info >= (3, 10, 0)
+
+SUPPORTS_FLATLITERAL = PY_310  # Literal[1, Literal[2]] -> Literal[1, 2] in Literal.__new__
+SUPPORTS_ISTYPEDDICT = PY_310  # typing.is_typeddict
+SUPPORTS_UNIONTYPE = PY_310  # types.UnionType
+
+PY_311: Final[bool] = sys.version_info >= (3, 11, 0)
+
+SUPPORTS_EXCEPTIONGROUP = PY_311  # ExceptionGroup
+SUPPORTS_EXCEPTIONNOTES = PY_311  # BaseException.__notes__
+SUPPORTS_NEVER = PY_311 or ("Never" in _typing_extensions__all__)  # typing.Never
+SUPPORTS_SELF = PY_311 or ("Self" in _typing_extensions__all__)  # typing.Self
+SUPPORTS_TOMLLIB = PY_311  # tomllib
+SUPPORTS_TYPEDDICTREQUIREDNESS = PY_311 or ("NotRequired" in _typing_extensions__all__ and "Required" in _typing_extensions__all__)  # NotRequired[T], Required[T]
+
+PY_312: Final[bool] = sys.version_info >= (3, 12, 0)
+
+SUPPORTS_MOREORIGBASES = PY_312  # python/cpython@0056701
+SUPPORTS_SYSMONITORING = PY_312  # sys.monitoring
+SUPPORTS_TYPEKEYWORD = PY_312  # type T
+SUPPORTS_WARNINGSKIPS = PY_312
+
+
+__all__ = [
+    "PY_39",
+    "SUPPORTS_ANNOTATED",
+    "SUPPORTS_GENERICBUILTINS",
+    "SUPPORTS_ZONEINFO",
+    "PY_310",
+    "SUPPORTS_FLATLITERAL",
+    "SUPPORTS_ISTYPEDDICT",
+    "SUPPORTS_UNIONTYPE",
+    "PY_311",
+    "SUPPORTS_EXCEPTIONGROUP",
+    "SUPPORTS_EXCEPTIONNOTES",
+    "SUPPORTS_NEVER",
+    "SUPPORTS_SELF",
+    "SUPPORTS_TOMLLIB",
+    "SUPPORTS_TYPEDDICTREQUIREDNESS",
+    "PY_312",
+    "SUPPORTS_MOREORIGBASES",
+    "SUPPORTS_SYSMONITORING",
+    "SUPPORTS_TYPEKEYWORD",
+    "SUPPORTS_WARNINGSKIPS",
+]