Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_descriptor.py: 57%

1"""A minimal native lock over a caller-owned file descriptor, contending with :class:`FileLock` on the same file.""" 

2 

3from __future__ import annotations 

4 

5__lazy_modules__ = ["filelock._unix", "threading", "typing"] 

6 

7import sys 

8import time 

9from threading import TIMEOUT_MAX 

10from typing import Final 

11 

12from filelock._api import _seconds 

13 

14if sys.platform == "win32": # pragma: win32 cover 

15 from filelock._windows import _lock_fd_nonblocking, _unlock_fd 

16else: # pragma: win32 no cover 

17 from filelock._unix import _lock_fd_nonblocking, _unlock_fd 

18 

19 

20def lock_descriptor(fd: int, *, blocking: bool = True, poll_interval: float = 0.05) -> bool: 

21 """ 

22 Take the native OS lock on *fd*, a file descriptor the caller opened and owns. 

23 

24 This is the same exclusive lock the native :class:`FileLock` backend uses: a whole-file :func:`fcntl.flock` lock 

25 on Unix, or a ``LockFileEx`` lock of one byte at offset 0 on Windows. A descriptor lock and a native path lock on 

26 the same file contend with each other. There are no offset or length parameters; seeking *fd* does not change 

27 the locked range. Interoperating with another tool requires agreement on the locking primitive, range, and protocol. 

28 

29 Unlike :class:`FileLock` it adds no path handling: it never opens, truncates, closes, unlinks, chmods, 

30 canonicalizes, or falls back. The caller owns *fd* before, during, and after the call, and must close it. 

31 On Windows *fd* must be a synchronous descriptor (its handle not opened with ``FILE_FLAG_OVERLAPPED``). 

32 

33 For timeout, reentrancy, singleton, lifetime, or stale-break behavior, use :class:`FileLock`. There is no async 

34 wrapper. Run this in an executor, or drive ``blocking=False`` from your own polling loop. 

35 

36 :param fd: an open file descriptor the caller owns. 

37 :param blocking: when ``True`` (default), retry the nonblocking attempt every *poll_interval* seconds until it 

38 succeeds; when ``False``, make one attempt. 

39 :param poll_interval: positive seconds between attempts while blocking, at most :data:`threading.TIMEOUT_MAX`; 

40 ignored when *blocking* is ``False``. 

41 

42 :returns: ``True`` once the lock is held, or ``False`` on contention when ``blocking`` is ``False``. 

43 

44 :raises OSError: for a permanent native failure, such as an invalid descriptor, or with ``errno.ENOSYS`` when the 

45 Python build lacks the native locking primitive. The descriptor is left open. 

46 :raises ValueError: if a blocking call receives a *poll_interval* outside that range, including ``nan``. 

47 

48 .. versionadded:: 3.30.0 

49 

50 """ 

51 if not blocking: 

52 return _lock_fd_nonblocking(fd) 

53 # time.sleep raises OverflowError past TIMEOUT_MAX; nan fails both comparisons. 

54 if not 0 < (poll_interval := _seconds("poll_interval", poll_interval)) <= TIMEOUT_MAX: 

55 msg: Final[str] = f"poll_interval must be greater than 0 and at most {TIMEOUT_MAX}, got {poll_interval}" 

56 raise ValueError(msg) 

57 while not _lock_fd_nonblocking(fd): 

58 time.sleep(poll_interval) 

59 return True 

60 

61 

62def unlock_descriptor(fd: int) -> None: 

63 """ 

64 Release the native OS lock on *fd* without touching the descriptor. 

65 

66 :param fd: the descriptor a prior :func:`lock_descriptor` locked; the caller still owns and must close it. 

67 

68 :raises OSError: if the native unlock fails, including ``errno.ENOSYS`` when the Python build lacks the native 

69 locking primitive; the caller may retry on the same descriptor. 

70 

71 .. versionadded:: 3.30.0 

72 

73 """ 

74 _unlock_fd(fd) 

75 

76 

77__all__ = [ 

78 "lock_descriptor", 

79 "unlock_descriptor", 

80]