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

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 

5import sys 

6import time 

7from math import isfinite 

8from typing import Final 

9 

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

11 from ._windows import _lock_fd_nonblocking, _unlock_fd 

12else: # pragma: win32 no cover 

13 from ._unix import _lock_fd_nonblocking, _unlock_fd 

14 

15 

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

17 """ 

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

19 

20 This is the same one-byte exclusive lock :class:`FileLock` uses, so a descriptor lock and a path lock on the same 

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

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

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

24 

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

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

27 

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

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

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

31 :param poll_interval: finite, positive seconds between attempts while blocking; ignored when *blocking* is 

32 ``False``. 

33 

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

35 

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

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

38 :raises ValueError: if a blocking call receives a non-finite or non-positive *poll_interval*. 

39 

40 .. versionadded:: 3.30.0 

41 

42 """ 

43 if not blocking: 

44 return _lock_fd_nonblocking(fd) 

45 if not isfinite(poll_interval) or poll_interval <= 0: 

46 msg: Final[str] = f"poll_interval must be finite and greater than 0, got {poll_interval}" 

47 raise ValueError(msg) 

48 while not _lock_fd_nonblocking(fd): 

49 time.sleep(poll_interval) 

50 return True 

51 

52 

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

54 """ 

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

56 

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

58 

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

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

61 

62 .. versionadded:: 3.30.0 

63 

64 """ 

65 _unlock_fd(fd) 

66 

67 

68__all__ = [ 

69 "lock_descriptor", 

70 "unlock_descriptor", 

71]