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