Use a metaclass to implement the singleton pattern (#340)

* add test to check the number of `__init__` calls

* FileLockMeta

* fix lint

* minor touch

* minor touch

* revert self._context

* fix type check

src/filelock/_api.py

@@ -5,10 +5,10 @@
 import os
 import time
 import warnings
-from abc import ABC, abstractmethod
+from abc import ABCMeta, abstractmethod
 from dataclasses import dataclass
 from threading import local
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 from weakref import WeakValueDictionary
 
 from ._error import Timeout
@@ -77,33 +77,63 @@ class ThreadLocalFileContext(FileLockContext, local):
     """A thread local version of the ``FileLockContext`` class."""
 
 
-class BaseFileLock(ABC, contextlib.ContextDecorator):
-    """Abstract base class for a file lock object."""
-
-    _instances: WeakValueDictionary[str, Self]
-
-    def __new__(  # noqa: PLR0913
+class FileLockMeta(ABCMeta):
+    def __call__(  # noqa: PLR0913
         cls,
         lock_file: str | os.PathLike[str],
-        timeout: float = -1,  # noqa: ARG003
-        mode: int = 0o644,  # noqa: ARG003
-        thread_local: bool = True,  # noqa: FBT001, FBT002, ARG003
+        timeout: float = -1,
+        mode: int = 0o644,
+        thread_local: bool = True,  # noqa: FBT001, FBT002
         *,
-        blocking: bool = True,  # noqa: ARG003
+        blocking: bool = True,
         is_singleton: bool = False,
-        **kwargs: Any,  # capture remaining kwargs for subclasses  # noqa: ARG003, ANN401
-    ) -> Self:
-        """Create a new lock object or if specified return the singleton instance for the lock file."""
-        if not is_singleton:
-            return super().__new__(cls)
-
-        instance = cls._instances.get(str(lock_file))
-        if not instance:
-            self = super().__new__(cls)
-            cls._instances[str(lock_file)] = self
-            return self
+        **kwargs: Any,  # capture remaining kwargs for subclasses  # noqa: ANN401
+    ) -> BaseFileLock:
+        if is_singleton:
+            instance = cls._instances.get(str(lock_file))  # type: ignore[attr-defined]
+            if instance:
+                params_to_check = {
+                    "thread_local": (thread_local, instance.is_thread_local()),
+                    "timeout": (timeout, instance.timeout),
+                    "mode": (mode, instance.mode),
+                    "blocking": (blocking, instance.blocking),
+                }
+
+                non_matching_params = {
+                    name: (passed_param, set_param)
+                    for name, (passed_param, set_param) in params_to_check.items()
+                    if passed_param != set_param
+                }
+                if not non_matching_params:
+                    return cast(BaseFileLock, instance)
+
+                # parameters do not match; raise error
+                msg = "Singleton lock instances cannot be initialized with differing arguments"
+                msg += "\nNon-matching arguments: "
+                for param_name, (passed_param, set_param) in non_matching_params.items():
+                    msg += f"\n\t{param_name} (existing lock has {set_param} but {passed_param} was passed)"
+                raise ValueError(msg)
+
+        instance = super().__call__(
+            lock_file=lock_file,
+            timeout=timeout,
+            mode=mode,
+            thread_local=thread_local,
+            blocking=blocking,
+            is_singleton=is_singleton,
+            **kwargs,
+        )
+
+        if is_singleton:
+            cls._instances[str(lock_file)] = instance  # type: ignore[attr-defined]
+
+        return cast(BaseFileLock, instance)
+
+
+class BaseFileLock(contextlib.ContextDecorator, metaclass=FileLockMeta):
+    """Abstract base class for a file lock object."""
 
-        return instance  # type: ignore[return-value] # https://github.com/python/mypy/issues/15322
+    _instances: WeakValueDictionary[str, BaseFileLock]
 
     def __init_subclass__(cls, **kwargs: dict[str, Any]) -> None:
         """Setup unique state for lock subclasses."""
@@ -136,34 +166,6 @@ def __init__(  # noqa: PLR0913
             to pass the same object around.
 
         """
-        if is_singleton and hasattr(self, "_context"):
-            # test whether other parameters match existing instance.
-            if not self.is_singleton:
-                msg = "__init__ should only be called on initialized object if it is a singleton"
-                raise RuntimeError(msg)
-
-            params_to_check = {
-                "thread_local": (thread_local, self.is_thread_local()),
-                "timeout": (timeout, self.timeout),
-                "mode": (mode, self.mode),
-                "blocking": (blocking, self.blocking),
-            }
-
-            non_matching_params = {
-                name: (passed_param, set_param)
-                for name, (passed_param, set_param) in params_to_check.items()
-                if passed_param != set_param
-            }
-            if not non_matching_params:
-                return  # bypass initialization because object is already initialized
-
-            # parameters do not match; raise error
-            msg = "Singleton lock instances cannot be initialized with differing arguments"
-            msg += "\nNon-matching arguments: "
-            for param_name, (passed_param, set_param) in non_matching_params.items():
-                msg += f"\n\t{param_name} (existing lock has {set_param} but {passed_param} was passed)"
-            raise ValueError(msg)
-
         self._is_thread_local = thread_local
         self._is_singleton = is_singleton
 

src/filelock/asyncio.py

@@ -9,9 +9,9 @@
 import time
 from dataclasses import dataclass
 from threading import local
-from typing import TYPE_CHECKING, Any, Callable, NoReturn
+from typing import TYPE_CHECKING, Any, Callable, NoReturn, cast
 
-from ._api import BaseFileLock, FileLockContext
+from ._api import BaseFileLock, FileLockContext, FileLockMeta
 from ._error import Timeout
 from ._soft import SoftFileLock
 from ._unix import UnixFileLock
@@ -67,7 +67,38 @@ async def __aexit__(  # noqa: D105
         await self.lock.release()
 
 
-class BaseAsyncFileLock(BaseFileLock):
+class AsyncFileLockMeta(FileLockMeta):
+    def __call__(  # type: ignore[override] # noqa: PLR0913
+        cls,  # noqa: N805
+        lock_file: str | os.PathLike[str],
+        timeout: float = -1,
+        mode: int = 0o644,
+        thread_local: bool = False,  # noqa: FBT001, FBT002
+        *,
+        blocking: bool = True,
+        is_singleton: bool = False,
+        loop: asyncio.AbstractEventLoop | None = None,
+        run_in_executor: bool = True,
+        executor: futures.Executor | None = None,
+    ) -> BaseAsyncFileLock:
+        if thread_local and run_in_executor:
+            msg = "run_in_executor is not supported when thread_local is True"
+            raise ValueError(msg)
+        instance = super().__call__(
+            lock_file=lock_file,
+            timeout=timeout,
+            mode=mode,
+            thread_local=thread_local,
+            blocking=blocking,
+            is_singleton=is_singleton,
+            loop=loop,
+            run_in_executor=run_in_executor,
+            executor=executor,
+        )
+        return cast(BaseAsyncFileLock, instance)
+
+
+class BaseAsyncFileLock(BaseFileLock, metaclass=AsyncFileLockMeta):
     """Base class for asynchronous file locks."""
 
     def __init__(  # noqa: PLR0913
@@ -104,9 +135,6 @@ def __init__(  # noqa: PLR0913
         """
         self._is_thread_local = thread_local
         self._is_singleton = is_singleton
-        if thread_local and run_in_executor:
-            msg = "run_in_executor is not supported when thread_local is True"
-            raise ValueError(msg)
 
         # Create the context. Note that external code should not work with the context directly and should instead use
         # properties of this class.

tests/test_filelock.py

@@ -785,3 +785,20 @@ class Lock2(lock_type):  # type: ignore[valid-type, misc]
     assert isinstance(Lock1._instances, WeakValueDictionary)  # noqa: SLF001
     assert isinstance(Lock2._instances, WeakValueDictionary)  # noqa: SLF001
     assert Lock1._instances is not Lock2._instances  # noqa: SLF001
+
+
+def test_singleton_locks_when_inheriting_init_is_called_once(tmp_path: Path) -> None:
+    init_calls = 0
+
+    class MyFileLock(FileLock):
+        def __init__(self, *args: Any, **kwargs: Any) -> None:  # noqa: ANN401
+            super().__init__(*args, **kwargs)
+            nonlocal init_calls
+            init_calls += 1
+
+    lock_path = tmp_path / "a"
+    lock1 = MyFileLock(str(lock_path), is_singleton=True)
+    lock2 = MyFileLock(str(lock_path), is_singleton=True)
+
+    assert lock1 is lock2
+    assert init_calls == 1