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 one-byte exclusive lock :class:`FileLock` uses, so a descriptor lock and a path lock on the same 

25 file contend with each other. Unlike :class:`FileLock` it adds no path handling: it never opens, truncates, closes, 

26 unlinks, chmods, canonicalizes, or falls back. The caller owns *fd* before, during, and after the call, and must 

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

28 

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

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

31 

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

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

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

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

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

37 

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

39 

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

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

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

43 

44 .. versionadded:: 3.30.0 

45 

46 """ 

47 if not blocking: 

48 return _lock_fd_nonblocking(fd) 

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

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

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

52 raise ValueError(msg) 

53 while not _lock_fd_nonblocking(fd): 

54 time.sleep(poll_interval) 

55 return True 

56 

57 

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

59 """ 

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

61 

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

63 

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

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

66 

67 .. versionadded:: 3.30.0 

68 

69 """ 

70 _unlock_fd(fd) 

71 

72 

73__all__ = [ 

74 "lock_descriptor", 

75 "unlock_descriptor", 

76]