singleton should be scoped to event loop

tests/test_search_provider.py

@@ -3,15 +3,7 @@
 from yente import settings
 
 from yente.exc import YenteIndexError, YenteNotFoundError
-from yente.provider import SearchProvider, ElasticSearchProvider, OpenSearchProvider
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("provider", [ElasticSearchProvider, OpenSearchProvider])
-async def test_is_singleton(provider):
-    s1 = await provider.create()
-    s2 = await provider.create()
-    assert id(s1) == id(s2)
+from yente.provider import SearchProvider
 
 
 @pytest.mark.asyncio

yente/provider/base.py

@@ -1,20 +1,21 @@
-from abc import abstractmethod
 from typing import Any, Dict, List, Optional
 from typing import AsyncIterator
 from threading import Lock
+from asyncio import get_running_loop
 
 
 class SingletonMeta(type):
     # This is a thread-safe implementation of the Singleton pattern. Shamelessly stolen from
     # https://refactoring.guru/design-patterns/singleton/python/example#example-1
-    _instances: Dict[type, Any] = {}
+    _instances: Dict[str, Any] = {}
     _lock: Lock = Lock()
 
     def __call__(cls, *args: Any, **kwds: Any) -> Any:
         with cls._lock:
-            if cls not in cls._instances:
-                cls._instances[cls] = super().__call__(*args, **kwds)
-        return cls._instances[cls]
+            loop_class = str(hash(get_running_loop())) + str(cls)
+            if loop_class not in cls._instances:
+                cls._instances[loop_class] = super().__call__(*args, **kwds)
+        return cls._instances[loop_class]
 
 
 class SearchProvider(object, metaclass=SingletonMeta):