|
1 | | -def singleton(myClass): |
2 | | - instances = {} |
| 1 | +from typing import Any, Generic, TypeVar |
3 | 2 |
|
4 | | - def getInstance(*args, **kwargs): |
5 | | - if myClass not in instances: |
6 | | - instances[myClass] = myClass(*args, **kwargs) |
7 | | - return instances[myClass] |
| 3 | +_T = TypeVar("_T") |
8 | 4 |
|
9 | | - # Expose the original class so callers can create fresh non-singleton instances |
10 | | - # when needed (e.g. per-chain ObjectsBucket in the fuzzer): |
11 | | - # fresh_bucket = ObjectsBucket.__wrapped__(api) |
12 | | - setattr(getInstance, "__wrapped__", myClass) |
13 | 5 |
|
14 | | - # Allow clearing the cached instance (useful for test isolation). |
15 | | - setattr(getInstance, "reset", lambda: instances.pop(myClass, None)) |
| 6 | +class _SingletonCallable(Generic[_T]): |
| 7 | + """Callable wrapper returned by the @singleton decorator. |
16 | 8 |
|
17 | | - return getInstance |
| 9 | + Exposes the original class as ``__wrapped__`` and provides a ``reset()`` |
| 10 | + helper that clears the cached instance (useful for test isolation). |
| 11 | + """ |
| 12 | + |
| 13 | + def __init__(self, cls: type[_T]) -> None: |
| 14 | + self.__wrapped__: type[_T] = cls |
| 15 | + self._instances: dict[type[_T], _T] = {} |
| 16 | + |
| 17 | + def __call__(self, *args: Any, **kwargs: Any) -> _T: |
| 18 | + if self.__wrapped__ not in self._instances: |
| 19 | + self._instances[self.__wrapped__] = self.__wrapped__(*args, **kwargs) |
| 20 | + return self._instances[self.__wrapped__] |
| 21 | + |
| 22 | + def reset(self) -> None: |
| 23 | + """Clear the cached instance so the next call creates a fresh one.""" |
| 24 | + self._instances.pop(self.__wrapped__, None) |
| 25 | + |
| 26 | + |
| 27 | +def singleton(myClass: type[_T]) -> _SingletonCallable[_T]: |
| 28 | + return _SingletonCallable(myClass) |
0 commit comments