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

1from __future__ import annotations 

2 

3import os 

4import sys 

5import warnings 

6from contextlib import suppress 

7from errno import EAGAIN, ENOSYS, EWOULDBLOCK 

8from pathlib import Path 

9from typing import cast 

10 

11from ._api import BaseFileLock 

12from ._util import ensure_directory_exists 

13 

14#: a flag to indicate if the fcntl API is available 

15has_fcntl = False 

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

17 

18 class UnixFileLock(BaseFileLock): 

19 """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems.""" 

20 

21 def _acquire(self) -> None: 

22 raise NotImplementedError 

23 

24 def _release(self) -> None: 

25 raise NotImplementedError 

26 

27else: # pragma: win32 no cover 

28 try: 

29 import fcntl 

30 

31 _ = (fcntl.flock, fcntl.LOCK_EX, fcntl.LOCK_NB, fcntl.LOCK_UN) 

32 except (ImportError, AttributeError): 

33 pass 

34 else: 

35 has_fcntl = True 

36 

37 class UnixFileLock(BaseFileLock): 

38 """ 

39 Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems. 

40 

41 Lock file cleanup: Unix and macOS delete the lock file reliably after release, even in 

42 multi-threaded scenarios. Unlike Windows, Unix allows unlinking files that other processes 

43 have open. 

44 """ 

45 

46 def _acquire(self) -> None: # noqa: C901, PLR0912 

47 ensure_directory_exists(self.lock_file) 

48 open_flags = os.O_RDWR | os.O_TRUNC 

49 o_nofollow = getattr(os, "O_NOFOLLOW", None) 

50 if o_nofollow is not None: 

51 open_flags |= o_nofollow 

52 open_flags |= os.O_CREAT 

53 open_mode = self._open_mode() 

54 try: 

55 fd = os.open(self.lock_file, open_flags, open_mode) 

56 except FileNotFoundError: 

57 # On FUSE/NFS, os.open(O_CREAT) is not atomic: LOOKUP + CREATE can be split, allowing a concurrent 

58 # unlink() to delete the file between them. For valid paths, treat ENOENT as transient contention. 

59 # For invalid paths (e.g., empty string), re-raise to avoid infinite retry loops. 

60 if self.lock_file and Path(self.lock_file).parent.exists(): 

61 return 

62 raise 

63 except PermissionError: 

64 # Sticky-bit dirs (e.g. /tmp): O_CREAT fails if the file is owned by another user (#317). 

65 # Fall back to opening the existing file without O_CREAT. 

66 if not Path(self.lock_file).exists(): 

67 raise 

68 try: 

69 fd = os.open(self.lock_file, open_flags & ~os.O_CREAT, open_mode) 

70 except FileNotFoundError: 

71 return 

72 if self.has_explicit_mode: 

73 with suppress(PermissionError): 

74 os.fchmod(fd, self._context.mode) 

75 try: 

76 fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) 

77 except OSError as exception: 

78 os.close(fd) 

79 if exception.errno == ENOSYS: 

80 with suppress(OSError): 

81 Path(self.lock_file).unlink() 

82 self._fallback_to_soft_lock() 

83 self._acquire() 

84 return 

85 if exception.errno not in {EAGAIN, EWOULDBLOCK}: 

86 raise 

87 else: 

88 # The file may have been unlinked by a concurrent _release() between our open() and flock(). 

89 # A lock on an unlinked inode is useless — discard and let the retry loop start fresh. 

90 if os.fstat(fd).st_nlink == 0: 

91 os.close(fd) 

92 else: 

93 self._context.lock_file_fd = fd 

94 

95 def _fallback_to_soft_lock(self) -> None: 

96 from ._soft import SoftFileLock # noqa: PLC0415 

97 

98 warnings.warn("flock not supported on this filesystem, falling back to SoftFileLock", stacklevel=2) 

99 from .asyncio import AsyncSoftFileLock, BaseAsyncFileLock # noqa: PLC0415 

100 

101 self.__class__ = AsyncSoftFileLock if isinstance(self, BaseAsyncFileLock) else SoftFileLock 

102 

103 def _release(self) -> None: 

104 fd = cast("int", self._context.lock_file_fd) 

105 self._context.lock_file_fd = None 

106 with suppress(OSError): 

107 Path(self.lock_file).unlink() 

108 fcntl.flock(fd, fcntl.LOCK_UN) 

109 with suppress(OSError): # close can raise EIO on FUSE/Docker bind-mount filesystems after unlink 

110 os.close(fd) 

111 

112 

113__all__ = [ 

114 "UnixFileLock", 

115 "has_fcntl", 

116]