make provider singleton

tests/test_search_provider.py

@@ -1,8 +1,17 @@
+# mypy: ignore-errors
 import pytest
 from yente import settings
 
 from yente.exc import YenteIndexError, YenteNotFoundError
-from yente.provider import SearchProvider
+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)
 
 
 @pytest.mark.asyncio

yente/provider/base.py

@@ -1,8 +1,22 @@
 from typing import Any, Dict, List, Optional
 from typing import AsyncIterator
+from threading import Lock
 
 
-class SearchProvider(object):
+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] = {}
+    _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]
+
+
+class SearchProvider(object, metaclass=SingletonMeta):
     async def close(self) -> None:
         raise NotImplementedError