47@typing.final
 48class Lazy(typing.Generic[T]):
 49    """A thread-safe value that gets lazily initialized at first access.
 50
 51    This type is thread-safe and may be called from multiple threads since this value
 52    can be initialized from multiple threads, however, any calls to `Lazy.get` will block
 53    if another thread is initializing it.
 54
 55    Example
 56    -------
 57    ```py
 58    # some expensive call that returns a `str`
 59    def expensive_string() -> str:
 60        return "hehehe"
 61
 62    STRING: Lazy[str] = Lazy(expensive_string)
 63    print(STRING.get()) # The string is built, stored in the lazy lock and returned.
 64    print(STRING.get()) # The string is retrieved from the lazy lock.
 65    ```
 66    """
 67
 68    __slots__ = ("__inner", "__lock")
 69
 70    def __init__(self, f: collections.Callable[[], T]) -> None:
 71        self.__inner: T | collections.Callable[[], T] = f
 72        self.__lock: threading.Lock | None = None
 73
 74    def get(self) -> T:
 75        """Get the value if it was initialized, otherwise initialize it and return it.
 76
 77        Its guaranteed to not block if the value has been initialized.
 78
 79        Example
 80        -------
 81        ```py
 82        # some expensive call that returns a `str`
 83        def expensive_string() -> str:
 84            return "hehehe"
 85
 86        STRING: Lazy[str] = Lazy(expensive_string)
 87        print(STRING.get()) # The string is built, stored in the lazy lock and returned.
 88        print(STRING.get()) # The string is retrieved from the lazy lock.
 89        ```
 90        """
 91        if not callable(self.__inner):
 92            # value is already initialized, no need to make a call.
 93            return self.__inner
 94
 95        if self.__lock is None:
 96            self.__lock = threading.Lock()
 97
 98        with self.__lock:
 99            # TYPE SAFETY: We know we need to call this function.
100            self.__inner = self.__inner()  # type: ignore
101            return self.__inner  # type: ignore
102
103    def __repr__(self) -> str:
104        if callable(self.__inner):
105            return "Lazy(uninit)"
106
107        return f"Lazy(value: {self.__inner.__repr__()})"
108
109    __str__ = __repr__
110
111    def __eq__(self, other: object) -> bool:
112        if not callable(self.__inner):
113            return False
114
115        if isinstance(other, Lazy):
116            return self.__inner == other.get()
117
118        return NotImplemented
119
120    def __ne__(self, other: object) -> bool:
121        return not self.__eq__(other)

 54@typing.final
 55class Once(typing.Generic[T]):
 56    """A synchronization primitive which can be written to only once.
 57
 58    Example
 59    -------
 60    ```py
 61    from sain.once import Once
 62    from uuid import uuid4, UUID
 63
 64    UUID: Once[UUID] = Once()
 65
 66    def start() -> None:
 67        assert UUID.get().is_none()
 68
 69        # First initialization.
 70        UUID.get_or(uuid4()) # some-uuid
 71
 72        # Won't set, returns the same uuid that got initialized first.
 73        UUID.get_or(uuid4()) # some-uuid
 74    ```
 75    """
 76
 77    __slots__ = ("_inner", "_lock")
 78
 79    def __init__(self) -> None:
 80        self._lock: threading.Lock | None = None
 81        self._inner: T | None = None
 82
 83    @property
 84    def is_set(self) -> bool:
 85        return self._inner is not None
 86
 87    def get(self) -> Option[T]:
 88        """Gets the stored value, returning `None` if not initialized.
 89
 90        This method will never block.
 91        """
 92        return _option.Some(self._inner) if self.is_set else _option.NOTHING  # pyright: ignore
 93
 94    @macros.unsafe
 95    def get_unchecked(self) -> T:
 96        """Get the contained value without checking if it was initialized.
 97
 98        Example
 99        -------
100        ```py
101        cell = Once[float]()
102        inner = cell.get_unchecked() # Undefined Behavior!!
103
104        # Initialize it first.
105        cell.get_or(math.sqrt(2.0))
106
107        # At this point of the program,
108        # it is guaranteed that the value is initialized.
109        inner = cell.get_unchecked()
110        ```
111        """
112        # SAFETY: The caller guarantees that the value is initialized.
113        return self.get().unwrap_unchecked()
114
115    def set(self, v: T) -> _result.Result[None, T]:
116        """Set the const value if its not set. returning `T` if its already set.
117
118        This method may block if another thread is trying to initialize the value.
119        The value is guaranteed to be set, just not necessarily the one provided.
120
121        Example
122        --------
123        ```py
124        flag = Once[bool]()
125        # flag is empty.
126        assert flag.get_or(True) is True.
127
128        # flag is not empty, so it returns the value we set first.
129        assert flag.set(False) == Err(True)
130        ```
131
132        Returns
133        -------
134        `sain.Result[None, T]`
135            This cell returns `Ok(None)` if it was empty. otherwise `Err(T)` if it was full.
136        """
137        if self._inner is not None:
138            return _result.Err(self._inner)
139
140        self._inner = self.get_or(v)
141        self._lock = None
142        return _result.Ok(None)
143
144    def clear(self) -> None:
145        """Clear the inner value, Setting it to `None`."""
146        self._lock = None
147        self._inner = None
148
149    def get_or(self, init: T) -> T:
150        """Get the value if it was not initialized, Otherwise set `init` value and returning it.
151
152        Many threads may call `get_or` concurrently with different
153        initializing functions, but it is guaranteed that only one function
154        will be executed.
155
156        Example
157        -------
158        ```py
159        UUID: Once[UUID] = Once()
160
161        def main() -> None:
162            assert UUID.get().is_none()
163
164            # First initialization.
165            UUID.get_or(uuid4()) # some-uuid
166
167            # Won't set, returns the same uuid that got initialized first.
168            UUID.get_or(uuid4()) # some-uuid
169            ```
170        """
171
172        # If the value is not empty we return it immediately.
173        if self._inner is not None:
174            return self._inner
175
176        if self._lock is None:
177            self._lock = threading.Lock()
178
179        with self._lock:
180            self._inner = init
181            return init
182
183    def get_or_with(self, f: collections.Callable[..., T]) -> T:
184        """Gets the contents of the cell, initializing it with `f` if the cell
185        was empty.
186
187        Many threads may call `get_or_with` concurrently with different
188        initializing functions, but it is guaranteed that only one function
189        will be executed.
190
191        Examples
192        --------
193        ```py
194        cell = Once[int]()
195        value = cell.get_or_with(lambda: 92)
196        assert value == 92
197
198        value = cell.get_or_with(lambda: 0)
199        assert value == 92
200        ```
201        """
202        # If the value is not empty we return it immediately.
203        if self._inner is not None:
204            return self._inner
205
206        if self._lock is None:
207            self._lock = threading.Lock()
208
209        with self._lock:
210            v = f()
211            self._inner = v
212            return v
213
214    def __repr__(self) -> str:
215        if not self.is_set:
216            return "<uninit>"
217
218        return f"Once(inner: {self._inner!r})"
219
220    __str__ = __repr__
221
222    def __eq__(self, __value: object) -> bool:
223        if isinstance(__value, Once):
224            return self._inner == __value._inner  # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType]
225
226        return NotImplemented
227
228    def __ne__(self, __value: object) -> bool:
229        return not self.__eq__(__value)
230
231    def __bool__(self) -> bool:
232        return self.is_set

124@typing.final
125class LazyFuture(typing.Generic[T]):
126    """A thread-safe value that gets lazily initialized asynchronously at first access.
127
128    This type is thread-safe and may be called from multiple threads since this value
129    can be initialized from multiple threads, however, any calls to `Lazy.get` will block
130    if another thread is initializing it.
131
132    Example
133    -------
134    ```py
135    # some expensive call that returns a `str`
136    async def fetch_expensive_string(filtered: bool) -> str:
137        return "hehehe" if filtered else "whahaha"
138
139    STRING: LazyFuture[str] = LazyFuture(lambda: expensive_string(True))
140    print(await STRING.get()) # The string is built, stored in the lazy lock and returned.
141    print(await STRING.get()) # The string is retrieved from the lazy lock.
142    ```
143    """
144
145    __slots__ = ("__inner", "__lock")
146
147    def __init__(
148        self,
149        f: collections.Callable[[], collections.Coroutine[typing.Any, typing.Any, T]],
150    ) -> None:
151        self.__inner: (
152            T
153            | collections.Callable[[], collections.Coroutine[typing.Any, typing.Any, T]]
154        ) = f
155        self.__lock: asyncio.Lock | None = asyncio.Lock()
156
157    async def get(self) -> T:
158        """Get the value if it was initialized, otherwise initialize it and return it.
159
160        Example
161        -------
162        ```py
163        # some expensive call that returns a `str`
164        async def fetch_expensive_string(filtered: bool) -> str:
165            return "hehehe" if filtered else "whahaha"
166
167        STRING: LazyFuture[str] = LazyFuture(lambda: expensive_string(True))
168        print(await STRING.get()) # The string is built, stored in the lazy lock and returned.
169        print(await STRING.get()) # The string is retrieved from the lazy lock.
170        ```
171        """
172        if not callable(self.__inner):
173            # value is already initialized, no need to make a call.
174            return self.__inner
175
176        if self.__lock is None:
177            self.__lock = asyncio.Lock()
178
179        async with self.__lock:
180            # calling self.__inner will make self.__inner type T
181            v = await self.__inner()  # pyright: ignore
182            self.__inner = v
183            return v
184
185    def __repr__(self) -> str:
186        if not callable(self.__inner):
187            return "LazyFuture(uninit)"
188
189        return f"LazyFuture(value: {self.__inner!r})"
190
191    __str__ = __repr__
192
193    def __eq__(self, other: object) -> bool:
194        if not callable(self.__inner):
195            return False
196
197        if isinstance(other, Lazy):
198            return self._inner == other._inner  # pyright: ignore
199
200        return NotImplemented
201
202    def __ne__(self, other: object) -> bool:
203        return not self.__eq__(other)

235@typing.final
236class AsyncOnce(typing.Generic[T]):
237    """A synchronization primitive which can be written to only once.
238
239    This is an `async` version of `Once`.
240
241    Example
242    -------
243    ```py
244    from sain.once import Once
245    from uuid import uuid4, UUID
246
247    # A global uuid
248    UUID: AsyncOnce[UUID] = AsyncOnce()
249
250    async def start() -> None:
251        assert UUID.get().is_none()
252        # First initialization.
253        await UUID.get_or(uuid4()) # some-uuid
254        # Won't set, returns the same uuid that got initialized first.
255        await UUID.get_or(uuid4()) # some-uuid
256    ```
257    """
258
259    __slots__ = ("_inner", "_lock")
260
261    def __init__(self) -> None:
262        self._lock: asyncio.Lock | None = None
263        self._inner: T | None = None
264
265    @property
266    def is_set(self) -> bool:
267        """Whether this inner value has ben initialized or not."""
268        return self._inner is not None
269
270    def get(self) -> Option[T]:
271        """Gets the stored value. `Some(None)` is returned if nothing is stored.
272
273        This method will never block.
274        """
275        return _option.Some(self._inner) if self.is_set else _option.NOTHING  # pyright: ignore
276
277    @macros.unsafe
278    def get_unchecked(self) -> T:
279        """Get the contained value without checking if it was initialized.
280
281        Example
282        -------
283        ```py
284        cell = AsyncOnce[float]()
285        inner = cell.get_unchecked() # Undefined Behavior!!
286
287        # Initialize it first.
288        cell.get_or(math.sqrt(2.0))
289
290        # At this point of the program,
291        # it is guaranteed that the value is initialized.
292        inner = cell.get_unchecked()
293        ```
294        """
295        # SAFETY: The caller guarantees that the value is initialized.
296        return self.get().unwrap_unchecked()
297
298    async def set(self, v: T) -> _result.Result[None, T]:
299        """Set the const value if its not set. returning `T` if its already set.
300
301        if another thread is trying to initialize the value, The value is guaranteed to be set,
302        just not necessarily the one provided.
303
304        Example
305        --------
306        ```py
307        flag = AsyncOnce[bool]()
308        # flag is empty.
309        assert await flag.get_or(True) is True.
310        # flag is not empty, so it returns the value we set first.
311        assert await flag.set(False) == Err(True)
312        ```
313
314        Returns
315        -------
316        `sain.Result[None, T]`
317            This cell returns `Ok(None)` if it was empty. otherwise `Err(T)` if it was full.
318        """
319        if self._inner is not None:
320            return _result.Err(self._inner)
321
322        self._inner = await self.get_or(v)
323        self._lock = None
324        return _result.Ok(None)
325
326    def clear(self) -> None:
327        """Clear the inner value, Setting it to `None`."""
328        self._lock = None
329        self._inner = None
330
331    async def get_or(self, init: T) -> T:
332        """Gets the contents of the cell, initializing it with `init` if the cell
333        was empty.
334
335        Many threads may call `get_or` concurrently with different
336        initializing functions, but it is guaranteed that only one function
337        will be executed.
338
339        Examples
340        --------
341        ```py
342        from sain.sync import AsyncOnce
343
344        cell = AsyncOnce[int]()
345        value = await cell.get_or(92)
346        assert value == 92
347
348        value = await cell.get_or(0)
349        assert value == 92
350        ```
351        """
352        # If the value is not empty we return it immediately.
353        if self._inner is not None:
354            return self._inner
355
356        if self._lock is None:
357            self._lock = asyncio.Lock()
358
359        async with self._lock:
360            self._inner = init
361            return init
362
363    async def get_or_with(self, f: collections.Callable[..., T]) -> T:
364        """Gets the contents of the cell, initializing it with `f` if the cell
365        was empty.
366
367        Many threads may call `get_or_with` concurrently with different
368        initializing functions, but it is guaranteed that only one function
369        will be executed.
370
371        Examples
372        --------
373        ```py
374        from sain.sync import AsyncOnce
375
376        cell = AsyncOnce[int]()
377        value = await cell.get_or_with(lambda: 92)
378        assert value == 92
379
380        value = await cell.get_or_init(lambda: 0)
381        assert value == 92
382        ```
383        """
384        # If the value is not empty we return it immediately.
385        if self._inner is not None:
386            return self._inner
387
388        if self._lock is None:
389            self._lock = asyncio.Lock()
390
391        async with self._lock:
392            v = f()
393            self._inner = v
394            return v
395
396    def __repr__(self) -> str:
397        if self._inner is not None:
398            return f"AsyncOnce(value: {self._inner})"
399        return "<async_uninit>"
400
401    __str__ = __repr__
402
403    def __eq__(self, __value: object) -> bool:
404        if isinstance(__value, AsyncOnce):
405            return self._inner == __value._inner  # pyright: ignore
406
407        return NotImplemented
408
409    def __ne__(self, __value: object) -> bool:
410        return not self.__eq__(__value)
411
412    def __bool__(self) -> bool:
413        return self.is_set