sain.boxed

  1# BSD 3-Clause License
  2#
  3# Copyright (c) 2022-Present, nxtlo
  4# All rights reserved.
  5#
  6# Redistribution and use in source and binary forms, with or without
  7# modification, are permitted provided that the following conditions are met:
  8#
  9# * Redistributions of source code must retain the above copyright notice, this
 10#   list of conditions and the following disclaimer.
 11#
 12# * Redistributions in binary form must reproduce the above copyright notice,
 13#   this list of conditions and the following disclaimer in the documentation
 14#   and/or other materials provided with the distribution.
 15#
 16# * Neither the name of the copyright holder nor the names of its
 17#   contributors may be used to endorse or promote products derived from
 18#   this software without specific prior written permission.
 19#
 20# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 21# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 22# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 23# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
 24# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 25# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 26# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 27# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 28# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 29# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 30
 31
 32"""a `Box` is a wrapper around a value that expires after the given amount of time."""
 33
 34from __future__ import annotations
 35
 36__all__ = ("Box",)
 37
 38import asyncio
 39import datetime
 40import math
 41import time
 42import typing
 43import warnings
 44
 45from sain.macros import ub_checks
 46
 47from . import futures
 48from . import option
 49
 50if typing.TYPE_CHECKING:
 51    from collections import abc as collections
 52
 53    from typing_extensions import Self
 54
 55    from . import Option
 56
 57T = typing.TypeVar("T", covariant=True)
 58
 59
 60@typing.final
 61class Box(typing.Generic[T]):
 62    """The box object for expiring data. not thread-safe.
 63
 64    A box is an object that contains a value of type `T` which expires it after the given amount of time,
 65    The box won't start expiring the data until its first access with `Box.get` method.
 66
 67    Example
 68    -------
 69    ```py
 70    # Initializing a box doesn't mean it started expiring. instead,
 71    # getting the value the first time will start the process.
 72    cache: dict[str, Box[int]] = {"sora": Box(999, timedelta(seconds=5)}
 73
 74    # first start expiring here.
 75    cache["sora"].get().unwrap()
 76    time.sleep(6)
 77    assert cache["sora"].has_expired()
 78    ```
 79    """
 80
 81    __slots__ = ("_inner", "_expire_in", "_on_expire", "_mono")
 82
 83    def __init__(self, value: T, expire_in: int | float | datetime.timedelta) -> None:
 84        if isinstance(expire_in, datetime.timedelta):
 85            expire_in = expire_in.total_seconds()
 86        else:
 87            expire_in = float(expire_in)
 88
 89        if expire_in <= 0:
 90            raise ValueError("expire_in must be more than 0 seconds.")
 91
 92        # We set the last call on the first access to the value.
 93        self._mono: float | None = None
 94        self._inner: Option[T] = option.Some(value)
 95        self._on_expire: collections.Callable[[T], typing.Any] | None = None
 96        self._expire_in = expire_in
 97
 98    @property
 99    def has_expired(self) -> bool:
100        """Returns True if the value has expired."""
101        # return self._mono is not None and not self._expire_in <= (
102        # time.monotonic() - self._mono
103        # )
104        return self._mono is not None and (
105            not self._mono or self._expire_in <= (time.monotonic() - self._mono)
106        )
107
108    def on_expire(self, callback: collections.Callable[[T], typing.Any]) -> Self:
109        """Set a callback that will be invoked when this value gets expired.
110
111        Both async and sync callbacks are supported.
112
113        Example
114        -------
115        ```py
116        async def sink(message: str) -> None:
117            await client.create_message(message)
118            print("Sinked", message)
119
120        box = Box("bluh", 5).on_expire(sink)
121
122        while box.get().is_some():
123            time.sleep(5)
124        ```
125        First `.get` call on an expired box, the `sink` callback will be invoked,
126        also the inner value will be set to `Some(None)`.
127
128        After 5 seconds.
129        ```py
130        assert box.get() == Some("bluh") # This last call invokes the callback.
131        # Sinked bluh
132        assert box.get().is_none()
133        ```
134        """
135        self._on_expire = callback
136        return self
137
138    def remaining(self) -> float:
139        """Returns when this box will expire in seconds.
140
141        Example
142        --------
143        ```py
144        jogo = Box("jogo", 3)
145        assert jogo.get().unwrap() == "jogo"
146
147        time.sleep(1)
148        assert jogo.remaining() == 2
149        ```
150        """
151        if not self._mono:
152            return 0.0
153
154        return math.floor(
155            (self._expire_in - (time.monotonic() - self._mono) + 1) * 0.99
156        )
157
158    def get(self) -> Option[T]:
159        """Get the contained value if it was not expired, otherwise `Some(None)` is returned.
160
161        Example
162        -------
163        ```py
164        pizza = Box("pizza", timedelta(days=1))
165
166        while not pizza.get().is_none():
167            # Do stuff with the value while its not expired.
168
169        # After 1 day.
170        assert pizza.get().is_none()
171        ```
172        """
173        if self.has_expired:
174            if self._on_expire is not None:
175                with warnings.catch_warnings():
176                    # ignore the warnings from `unwrap_unchecked`.
177                    warnings.simplefilter("ignore", category=ub_checks)
178                    try:
179                        if asyncio.iscoroutinefunction(self._on_expire):
180                            futures.loop().run_until_complete(
181                                self._on_expire(self._inner.unwrap_unchecked())
182                            )
183                        else:
184                            self._on_expire(self._inner.unwrap_unchecked())
185                    finally:
186                        self._on_expire = None
187
188            self._inner = option.NOTHING
189            self._mono = None
190            # SAFETY: The value is expired, therefore we always return None.
191            return option.NOTHING
192
193        if self._mono is None:
194            self._mono = time.monotonic()
195
196        return self._inner
197
198    def __repr__(self) -> str:
199        return f"Box(value: {self._inner}, expired: {self.has_expired})"
200
201    __str__ = __repr__
202
203    def __hash__(self) -> int:
204        return hash(self._inner)
205
206    def __bool__(self) -> bool:
207        return not self.has_expired