Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/asyncio.py: 41%
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
1"""An asyncio-based implementation of the file lock."""
3from __future__ import annotations
5import asyncio
6import contextlib
7import logging
8import os
9import time
10from dataclasses import dataclass
11from inspect import iscoroutinefunction
12from threading import local
13from typing import TYPE_CHECKING, Any, NoReturn, TypeVar
15from ._api import _UNSET_FILE_MODE, BaseFileLock, FileLockContext, FileLockMeta
16from ._error import Timeout
17from ._soft import SoftFileLock
18from ._unix import UnixFileLock
19from ._windows import WindowsFileLock
21if TYPE_CHECKING:
22 import sys
23 from collections.abc import Callable
24 from concurrent import futures
25 from types import TracebackType
27 if sys.version_info >= (3, 11): # pragma: no cover (py311+)
28 from typing import Self
29 else: # pragma: no cover (<py311)
30 from typing_extensions import Self
33_LOGGER = logging.getLogger("filelock")
36@dataclass
37class AsyncFileLockContext(FileLockContext):
38 """A dataclass which holds the context for a ``BaseAsyncFileLock`` object."""
40 #: Whether run in executor
41 run_in_executor: bool = True
43 #: The executor
44 executor: futures.Executor | None = None
46 #: The loop
47 loop: asyncio.AbstractEventLoop | None = None
50class AsyncThreadLocalFileContext(AsyncFileLockContext, local):
51 """A thread local version of the ``FileLockContext`` class."""
54class AsyncAcquireReturnProxy:
55 """A context-aware object that will release the lock file when exiting."""
57 def __init__(self, lock: BaseAsyncFileLock) -> None: # noqa: D107
58 self.lock = lock
60 async def __aenter__(self) -> BaseAsyncFileLock: # noqa: D105
61 return self.lock
63 async def __aexit__( # noqa: D105
64 self,
65 exc_type: type[BaseException] | None,
66 exc_value: BaseException | None,
67 traceback: TracebackType | None,
68 ) -> None:
69 await self.lock.release()
72_AT = TypeVar("_AT", bound="BaseAsyncFileLock")
75class AsyncFileLockMeta(FileLockMeta):
76 def __call__( # ty: ignore[invalid-method-override] # noqa: PLR0913
77 cls: type[_AT], # noqa: N805
78 lock_file: str | os.PathLike[str],
79 timeout: float = -1,
80 mode: int = _UNSET_FILE_MODE,
81 thread_local: bool = False, # noqa: FBT001, FBT002
82 *,
83 blocking: bool = True,
84 is_singleton: bool = False,
85 poll_interval: float = 0.05,
86 lifetime: float | None = None,
87 loop: asyncio.AbstractEventLoop | None = None,
88 run_in_executor: bool = True,
89 executor: futures.Executor | None = None,
90 ) -> _AT:
91 if thread_local and run_in_executor:
92 msg = "run_in_executor is not supported when thread_local is True"
93 raise ValueError(msg)
94 return super().__call__(
95 lock_file=lock_file,
96 timeout=timeout,
97 mode=mode,
98 thread_local=thread_local,
99 blocking=blocking,
100 is_singleton=is_singleton,
101 poll_interval=poll_interval,
102 lifetime=lifetime,
103 loop=loop,
104 run_in_executor=run_in_executor,
105 executor=executor,
106 )
109class BaseAsyncFileLock(BaseFileLock, metaclass=AsyncFileLockMeta):
110 """
111 Base class for asynchronous file locks.
113 .. versionadded:: 3.15.0
115 """
117 def __init__( # noqa: PLR0913
118 self,
119 lock_file: str | os.PathLike[str],
120 timeout: float = -1,
121 mode: int = _UNSET_FILE_MODE,
122 thread_local: bool = False, # noqa: FBT001, FBT002
123 *,
124 blocking: bool = True,
125 is_singleton: bool = False,
126 poll_interval: float = 0.05,
127 lifetime: float | None = None,
128 loop: asyncio.AbstractEventLoop | None = None,
129 run_in_executor: bool = True,
130 executor: futures.Executor | None = None,
131 ) -> None:
132 """
133 Create a new lock object.
135 :param lock_file: path to the file
136 :param timeout: default timeout when acquiring the lock, in seconds. It will be used as fallback value in the
137 acquire method, if no timeout value (``None``) is given. If you want to disable the timeout, set it to a
138 negative value. A timeout of 0 means that there is exactly one attempt to acquire the file lock.
139 :param mode: file permissions for the lockfile. When not specified, the OS controls permissions via umask and
140 default ACLs, preserving POSIX default ACL inheritance in shared directories.
141 :param thread_local: Whether this object's internal context should be thread local or not. If this is set to
142 ``False`` then the lock will be reentrant across threads. When ``True`` (the default), **all fields of the
143 lock's internal context are per-thread**, including the configuration values ``poll_interval``, ``timeout``,
144 ``blocking``, ``mode``, and ``lifetime``. Setting one of these properties from one thread does not change
145 the value seen by another thread; threads that did not perform the write continue to see the value supplied
146 at construction time. If you need configuration values to be visible across threads, construct the lock
147 with ``thread_local=False``.
148 :param blocking: whether the lock should be blocking or not
149 :param is_singleton: If this is set to ``True`` then only one instance of this class will be created per lock
150 file. This is useful if you want to use the lock object for reentrant locking without needing to pass the
151 same object around.
152 :param poll_interval: default interval for polling the lock file, in seconds. It will be used as fallback value
153 in the acquire method, if no poll_interval value (``None``) is given.
154 :param lifetime: maximum time in seconds a lock can be held before it is considered expired. When set, a waiting
155 process will break a lock whose file modification time is older than ``lifetime`` seconds. ``None`` (the
156 default) means locks never expire.
157 :param loop: The event loop to use. If not specified, the running event loop will be used.
158 :param run_in_executor: If this is set to ``True`` then the lock will be acquired in an executor.
159 :param executor: The executor to use. If not specified, the default executor will be used.
161 """
162 self._is_thread_local = thread_local
163 self._is_singleton = is_singleton
165 # Create the context. Note that external code should not work with the context directly and should instead use
166 # properties of this class.
167 kwargs: dict[str, Any] = {
168 "lock_file": os.fspath(lock_file),
169 "timeout": timeout,
170 "mode": mode,
171 "blocking": blocking,
172 "poll_interval": poll_interval,
173 "lifetime": lifetime,
174 "loop": loop,
175 "run_in_executor": run_in_executor,
176 "executor": executor,
177 }
178 self._context: AsyncFileLockContext = (AsyncThreadLocalFileContext if thread_local else AsyncFileLockContext)(
179 **kwargs
180 )
182 @property
183 def run_in_executor(self) -> bool:
184 """:returns: whether run in executor."""
185 return self._context.run_in_executor
187 @property
188 def executor(self) -> futures.Executor | None:
189 """:returns: the executor."""
190 return self._context.executor
192 @executor.setter
193 def executor(self, value: futures.Executor | None) -> None: # pragma: no cover
194 """
195 Change the executor.
197 :param futures.Executor | None value: the new executor or ``None``
199 """
200 self._context.executor = value
202 @property
203 def loop(self) -> asyncio.AbstractEventLoop | None:
204 """:returns: the event loop."""
205 return self._context.loop
207 async def acquire( # ty: ignore[invalid-method-override]
208 self,
209 timeout: float | None = None,
210 poll_interval: float | None = None,
211 *,
212 blocking: bool | None = None,
213 cancel_check: Callable[[], bool] | None = None,
214 ) -> AsyncAcquireReturnProxy:
215 """
216 Try to acquire the file lock.
218 :param timeout: maximum wait time for acquiring the lock, ``None`` means use the default
219 :attr:`~BaseFileLock.timeout` is and if ``timeout < 0``, there is no timeout and this method will block
220 until the lock could be acquired
221 :param poll_interval: interval of trying to acquire the lock file, ``None`` means use the default
222 :attr:`~BaseFileLock.poll_interval`
223 :param blocking: defaults to True. If False, function will return immediately if it cannot obtain a lock on the
224 first attempt. Otherwise, this method will block until the timeout expires or the lock is acquired.
225 :param cancel_check: a callable returning ``True`` when the acquisition should be canceled. Checked on each poll
226 iteration. When triggered, raises :class:`~Timeout` just like an expired timeout.
228 :returns: a context object that will unlock the file when the context is exited
230 :raises Timeout: if fails to acquire lock within the timeout period
232 .. code-block:: python
234 # You can use this method in the context manager (recommended)
235 with lock.acquire():
236 pass
238 # Or use an equivalent try-finally construct:
239 lock.acquire()
240 try:
241 pass
242 finally:
243 lock.release()
245 """
246 # Use the default timeout, if no timeout is provided.
247 if timeout is None:
248 timeout = self._context.timeout
250 if blocking is None:
251 blocking = self._context.blocking
253 if poll_interval is None:
254 poll_interval = self._context.poll_interval
256 # Increment the number right at the beginning. We can still undo it, if something fails.
257 self._context.lock_counter += 1
259 start_time = time.perf_counter()
260 try:
261 await self._async_poll_until_acquired(
262 blocking=blocking,
263 cancel_check=cancel_check,
264 timeout=timeout,
265 poll_interval=poll_interval,
266 start_time=start_time,
267 )
268 except BaseException: # Something did go wrong, so decrement the counter.
269 self._context.lock_counter = max(0, self._context.lock_counter - 1)
270 raise
271 return AsyncAcquireReturnProxy(lock=self)
273 async def _async_poll_until_acquired(
274 self,
275 *,
276 blocking: bool,
277 cancel_check: Callable[[], bool] | None,
278 timeout: float,
279 poll_interval: float,
280 start_time: float,
281 ) -> None:
282 lock_id = id(self)
283 lock_filename = self.lock_file
284 while True:
285 if not self.is_locked:
286 self._try_break_expired_lock()
287 _LOGGER.debug("Attempting to acquire lock %s on %s", lock_id, lock_filename)
288 await self._run_internal_method(self._acquire)
289 if self.is_locked:
290 _LOGGER.debug("Lock %s acquired on %s", lock_id, lock_filename)
291 return
292 if self._check_give_up(
293 lock_id,
294 lock_filename,
295 blocking=blocking,
296 cancel_check=cancel_check,
297 timeout=timeout,
298 start_time=start_time,
299 ):
300 raise Timeout(lock_filename)
301 msg = "Lock %s not acquired on %s, waiting %s seconds ..."
302 _LOGGER.debug(msg, lock_id, lock_filename, poll_interval)
303 await asyncio.sleep(poll_interval)
305 async def release(self, force: bool = False) -> None: # ty: ignore[invalid-method-override] # noqa: FBT001, FBT002
306 """
307 Release the file lock. The lock is only completely released when the lock counter reaches 0. The lock file
308 itself may be deleted automatically, the behavior is platform-specific.
310 :param force: If true, the lock counter is ignored and the lock is released in every case.
312 """
313 if self.is_locked:
314 self._context.lock_counter -= 1
316 if self._context.lock_counter == 0 or force:
317 lock_id, lock_filename = id(self), self.lock_file
319 _LOGGER.debug("Attempting to release lock %s on %s", lock_id, lock_filename)
320 await self._run_internal_method(self._release)
321 self._context.lock_counter = 0
322 _LOGGER.debug("Lock %s released on %s", lock_id, lock_filename)
324 async def _run_internal_method(self, method: Callable[[], Any]) -> None:
325 if iscoroutinefunction(method):
326 await method()
327 elif self.run_in_executor:
328 await asyncio.get_running_loop().run_in_executor(self.executor, method)
329 else:
330 method()
332 def __enter__(self) -> NoReturn:
333 """Sync context manager entry is not supported because lock acquisition is a coroutine."""
334 msg = "Use `async with` — acquire/release are coroutines and cannot be awaited in a sync context manager."
335 raise NotImplementedError(msg)
337 def __exit__(
338 self,
339 exc_type: type[BaseException] | None,
340 exc_value: BaseException | None,
341 traceback: object,
342 ) -> None:
343 """Sync context manager exit is not supported because lock release is a coroutine."""
344 msg = "Use `async with` — acquire/release are coroutines and cannot be awaited in a sync context manager."
345 raise NotImplementedError(msg)
347 async def __aenter__(self) -> Self:
348 """
349 Acquire the lock.
351 :returns: the lock object
353 """
354 await self.acquire()
355 return self
357 async def __aexit__(
358 self,
359 exc_type: type[BaseException] | None,
360 exc_value: BaseException | None,
361 traceback: TracebackType | None,
362 ) -> None:
363 """
364 Release the lock.
366 :param exc_type: the exception type if raised
367 :param exc_value: the exception value if raised
368 :param traceback: the exception traceback if raised
370 """
371 await self.release()
373 def __del__(self) -> None:
374 """Release on deletion — safe to call during GC even when no event loop is running."""
375 with contextlib.suppress(Exception):
376 try:
377 loop = asyncio.get_running_loop()
378 except RuntimeError:
379 # No running loop — try stored loop or create one
380 loop = self._context.loop if self._context.loop and not self._context.loop.is_closed() else None
381 if loop is None:
382 return
383 if not loop.is_running(): # pragma: no cover
384 loop.run_until_complete(self.release(force=True))
385 else:
386 loop.create_task(self.release(force=True))
389class AsyncSoftFileLock(SoftFileLock, BaseAsyncFileLock):
390 """Simply watches the existence of the lock file."""
393class AsyncUnixFileLock(UnixFileLock, BaseAsyncFileLock):
394 """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems."""
397class AsyncWindowsFileLock(WindowsFileLock, BaseAsyncFileLock):
398 """Uses the :func:`msvcrt.locking` to hard lock the lock file on windows systems."""
401__all__ = [
402 "AsyncAcquireReturnProxy",
403 "AsyncSoftFileLock",
404 "AsyncUnixFileLock",
405 "AsyncWindowsFileLock",
406 "BaseAsyncFileLock",
407]