Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/filelock/_api.py: 70%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import contextlib
4import inspect
5import logging
6import os
7import time
8import warnings
9from abc import ABCMeta, abstractmethod
10from dataclasses import dataclass
11from threading import local
12from typing import TYPE_CHECKING, Any, cast
13from weakref import WeakValueDictionary
15from ._error import Timeout
17if TYPE_CHECKING:
18 import sys
19 from types import TracebackType
21 if sys.version_info >= (3, 11): # pragma: no cover (py311+)
22 from typing import Self
23 else: # pragma: no cover (<py311)
24 from typing_extensions import Self
27_LOGGER = logging.getLogger("filelock")
30# This is a helper class which is returned by :meth:`BaseFileLock.acquire` and wraps the lock to make sure __enter__
31# is not called twice when entering the with statement. If we would simply return *self*, the lock would be acquired
32# again in the *__enter__* method of the BaseFileLock, but not released again automatically. issue #37 (memory leak)
33class AcquireReturnProxy:
34 """A context-aware object that will release the lock file when exiting."""
36 def __init__(self, lock: BaseFileLock) -> None:
37 self.lock = lock
39 def __enter__(self) -> BaseFileLock:
40 return self.lock
42 def __exit__(
43 self,
44 exc_type: type[BaseException] | None,
45 exc_value: BaseException | None,
46 traceback: TracebackType | None,
47 ) -> None:
48 self.lock.release()
51@dataclass
52class FileLockContext:
53 """A dataclass which holds the context for a ``BaseFileLock`` object."""
55 # The context is held in a separate class to allow optional use of thread local storage via the
56 # ThreadLocalFileContext class.
58 #: The path to the lock file.
59 lock_file: str
61 #: The default timeout value.
62 timeout: float
64 #: The mode for the lock files
65 mode: int
67 #: Whether the lock should be blocking or not
68 blocking: bool
70 #: The file descriptor for the *_lock_file* as it is returned by the os.open() function, not None when lock held
71 lock_file_fd: int | None = None
73 #: The lock counter is used for implementing the nested locking mechanism.
74 lock_counter: int = 0 # When the lock is acquired is increased and the lock is only released, when this value is 0
77class ThreadLocalFileContext(FileLockContext, local):
78 """A thread local version of the ``FileLockContext`` class."""
81class FileLockMeta(ABCMeta):
82 def __call__( # noqa: PLR0913
83 cls,
84 lock_file: str | os.PathLike[str],
85 timeout: float = -1,
86 mode: int = 0o644,
87 thread_local: bool = True, # noqa: FBT001, FBT002
88 *,
89 blocking: bool = True,
90 is_singleton: bool = False,
91 **kwargs: Any, # capture remaining kwargs for subclasses # noqa: ANN401
92 ) -> BaseFileLock:
93 if is_singleton:
94 instance = cls._instances.get(str(lock_file)) # type: ignore[attr-defined]
95 if instance:
96 params_to_check = {
97 "thread_local": (thread_local, instance.is_thread_local()),
98 "timeout": (timeout, instance.timeout),
99 "mode": (mode, instance.mode),
100 "blocking": (blocking, instance.blocking),
101 }
103 non_matching_params = {
104 name: (passed_param, set_param)
105 for name, (passed_param, set_param) in params_to_check.items()
106 if passed_param != set_param
107 }
108 if not non_matching_params:
109 return cast(BaseFileLock, instance)
111 # parameters do not match; raise error
112 msg = "Singleton lock instances cannot be initialized with differing arguments"
113 msg += "\nNon-matching arguments: "
114 for param_name, (passed_param, set_param) in non_matching_params.items():
115 msg += f"\n\t{param_name} (existing lock has {set_param} but {passed_param} was passed)"
116 raise ValueError(msg)
118 # Workaround to make `__init__`'s params optional in subclasses
119 # E.g. virtualenv changes the signature of the `__init__` method in the `BaseFileLock` class descendant
120 # (https://github.com/tox-dev/filelock/pull/340)
122 all_params = {
123 "timeout": timeout,
124 "mode": mode,
125 "thread_local": thread_local,
126 "blocking": blocking,
127 "is_singleton": is_singleton,
128 **kwargs,
129 }
131 present_params = inspect.signature(cls.__init__).parameters # type: ignore[misc]
132 init_params = {key: value for key, value in all_params.items() if key in present_params}
134 instance = super().__call__(lock_file, **init_params)
136 if is_singleton:
137 cls._instances[str(lock_file)] = instance # type: ignore[attr-defined]
139 return cast(BaseFileLock, instance)
142class BaseFileLock(contextlib.ContextDecorator, metaclass=FileLockMeta):
143 """Abstract base class for a file lock object."""
145 _instances: WeakValueDictionary[str, BaseFileLock]
147 def __init_subclass__(cls, **kwargs: dict[str, Any]) -> None:
148 """Setup unique state for lock subclasses."""
149 super().__init_subclass__(**kwargs)
150 cls._instances = WeakValueDictionary()
152 def __init__( # noqa: PLR0913
153 self,
154 lock_file: str | os.PathLike[str],
155 timeout: float = -1,
156 mode: int = 0o644,
157 thread_local: bool = True, # noqa: FBT001, FBT002
158 *,
159 blocking: bool = True,
160 is_singleton: bool = False,
161 ) -> None:
162 """
163 Create a new lock object.
165 :param lock_file: path to the file
166 :param timeout: default timeout when acquiring the lock, in seconds. It will be used as fallback value in \
167 the acquire method, if no timeout value (``None``) is given. If you want to disable the timeout, set it \
168 to a negative value. A timeout of 0 means that there is exactly one attempt to acquire the file lock.
169 :param mode: file permissions for the lockfile
170 :param thread_local: Whether this object's internal context should be thread local or not. If this is set to \
171 ``False`` then the lock will be reentrant across threads.
172 :param blocking: whether the lock should be blocking or not
173 :param is_singleton: If this is set to ``True`` then only one instance of this class will be created \
174 per lock file. This is useful if you want to use the lock object for reentrant locking without needing \
175 to pass the same object around.
177 """
178 self._is_thread_local = thread_local
179 self._is_singleton = is_singleton
181 # Create the context. Note that external code should not work with the context directly and should instead use
182 # properties of this class.
183 kwargs: dict[str, Any] = {
184 "lock_file": os.fspath(lock_file),
185 "timeout": timeout,
186 "mode": mode,
187 "blocking": blocking,
188 }
189 self._context: FileLockContext = (ThreadLocalFileContext if thread_local else FileLockContext)(**kwargs)
191 def is_thread_local(self) -> bool:
192 """:return: a flag indicating if this lock is thread local or not"""
193 return self._is_thread_local
195 @property
196 def is_singleton(self) -> bool:
197 """:return: a flag indicating if this lock is singleton or not"""
198 return self._is_singleton
200 @property
201 def lock_file(self) -> str:
202 """:return: path to the lock file"""
203 return self._context.lock_file
205 @property
206 def timeout(self) -> float:
207 """
208 :return: the default timeout value, in seconds
210 .. versionadded:: 2.0.0
211 """
212 return self._context.timeout
214 @timeout.setter
215 def timeout(self, value: float | str) -> None:
216 """
217 Change the default timeout value.
219 :param value: the new value, in seconds
221 """
222 self._context.timeout = float(value)
224 @property
225 def blocking(self) -> bool:
226 """:return: whether the locking is blocking or not"""
227 return self._context.blocking
229 @blocking.setter
230 def blocking(self, value: bool) -> None:
231 """
232 Change the default blocking value.
234 :param value: the new value as bool
236 """
237 self._context.blocking = value
239 @property
240 def mode(self) -> int:
241 """:return: the file permissions for the lockfile"""
242 return self._context.mode
244 @abstractmethod
245 def _acquire(self) -> None:
246 """If the file lock could be acquired, self._context.lock_file_fd holds the file descriptor of the lock file."""
247 raise NotImplementedError
249 @abstractmethod
250 def _release(self) -> None:
251 """Releases the lock and sets self._context.lock_file_fd to None."""
252 raise NotImplementedError
254 @property
255 def is_locked(self) -> bool:
256 """
258 :return: A boolean indicating if the lock file is holding the lock currently.
260 .. versionchanged:: 2.0.0
262 This was previously a method and is now a property.
263 """
264 return self._context.lock_file_fd is not None
266 @property
267 def lock_counter(self) -> int:
268 """:return: The number of times this lock has been acquired (but not yet released)."""
269 return self._context.lock_counter
271 def acquire(
272 self,
273 timeout: float | None = None,
274 poll_interval: float = 0.05,
275 *,
276 poll_intervall: float | None = None,
277 blocking: bool | None = None,
278 ) -> AcquireReturnProxy:
279 """
280 Try to acquire the file lock.
282 :param timeout: maximum wait time for acquiring the lock, ``None`` means use the default :attr:`~timeout` is and
283 if ``timeout < 0``, there is no timeout and this method will block until the lock could be acquired
284 :param poll_interval: interval of trying to acquire the lock file
285 :param poll_intervall: deprecated, kept for backwards compatibility, use ``poll_interval`` instead
286 :param blocking: defaults to True. If False, function will return immediately if it cannot obtain a lock on the
287 first attempt. Otherwise, this method will block until the timeout expires or the lock is acquired.
288 :raises Timeout: if fails to acquire lock within the timeout period
289 :return: a context object that will unlock the file when the context is exited
291 .. code-block:: python
293 # You can use this method in the context manager (recommended)
294 with lock.acquire():
295 pass
297 # Or use an equivalent try-finally construct:
298 lock.acquire()
299 try:
300 pass
301 finally:
302 lock.release()
304 .. versionchanged:: 2.0.0
306 This method returns now a *proxy* object instead of *self*,
307 so that it can be used in a with statement without side effects.
309 """
310 # Use the default timeout, if no timeout is provided.
311 if timeout is None:
312 timeout = self._context.timeout
314 if blocking is None:
315 blocking = self._context.blocking
317 if poll_intervall is not None:
318 msg = "use poll_interval instead of poll_intervall"
319 warnings.warn(msg, DeprecationWarning, stacklevel=2)
320 poll_interval = poll_intervall
322 # Increment the number right at the beginning. We can still undo it, if something fails.
323 self._context.lock_counter += 1
325 lock_id = id(self)
326 lock_filename = self.lock_file
327 start_time = time.perf_counter()
328 try:
329 while True:
330 if not self.is_locked:
331 _LOGGER.debug("Attempting to acquire lock %s on %s", lock_id, lock_filename)
332 self._acquire()
333 if self.is_locked:
334 _LOGGER.debug("Lock %s acquired on %s", lock_id, lock_filename)
335 break
336 if blocking is False:
337 _LOGGER.debug("Failed to immediately acquire lock %s on %s", lock_id, lock_filename)
338 raise Timeout(lock_filename) # noqa: TRY301
339 if 0 <= timeout < time.perf_counter() - start_time:
340 _LOGGER.debug("Timeout on acquiring lock %s on %s", lock_id, lock_filename)
341 raise Timeout(lock_filename) # noqa: TRY301
342 msg = "Lock %s not acquired on %s, waiting %s seconds ..."
343 _LOGGER.debug(msg, lock_id, lock_filename, poll_interval)
344 time.sleep(poll_interval)
345 except BaseException: # Something did go wrong, so decrement the counter.
346 self._context.lock_counter = max(0, self._context.lock_counter - 1)
347 raise
348 return AcquireReturnProxy(lock=self)
350 def release(self, force: bool = False) -> None: # noqa: FBT001, FBT002
351 """
352 Releases the file lock. Please note, that the lock is only completely released, if the lock counter is 0.
353 Also note, that the lock file itself is not automatically deleted.
355 :param force: If true, the lock counter is ignored and the lock is released in every case/
357 """
358 if self.is_locked:
359 self._context.lock_counter -= 1
361 if self._context.lock_counter == 0 or force:
362 lock_id, lock_filename = id(self), self.lock_file
364 _LOGGER.debug("Attempting to release lock %s on %s", lock_id, lock_filename)
365 self._release()
366 self._context.lock_counter = 0
367 _LOGGER.debug("Lock %s released on %s", lock_id, lock_filename)
369 def __enter__(self) -> Self:
370 """
371 Acquire the lock.
373 :return: the lock object
375 """
376 self.acquire()
377 return self
379 def __exit__(
380 self,
381 exc_type: type[BaseException] | None,
382 exc_value: BaseException | None,
383 traceback: TracebackType | None,
384 ) -> None:
385 """
386 Release the lock.
388 :param exc_type: the exception type if raised
389 :param exc_value: the exception value if raised
390 :param traceback: the exception traceback if raised
392 """
393 self.release()
395 def __del__(self) -> None:
396 """Called when the lock object is deleted."""
397 self.release(force=True)
400__all__ = [
401 "AcquireReturnProxy",
402 "BaseFileLock",
403]