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

1from __future__ import annotations 

2 

3import os 

4import sys 

5from contextlib import suppress 

6from errno import ENOSYS 

7from pathlib import Path 

8from typing import cast 

9 

10from ._api import BaseFileLock 

11from ._util import ensure_directory_exists 

12 

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

14has_fcntl = False 

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

16 

17 class UnixFileLock(BaseFileLock): 

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

19 

20 def _acquire(self) -> None: 

21 raise NotImplementedError 

22 

23 def _release(self) -> None: 

24 raise NotImplementedError 

25 

26else: # pragma: win32 no cover 

27 try: 

28 import fcntl 

29 

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

31 except (ImportError, AttributeError): 

32 pass 

33 else: 

34 has_fcntl = True 

35 

36 class UnixFileLock(BaseFileLock): 

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

38 

39 def _acquire(self) -> None: 

40 ensure_directory_exists(self.lock_file) 

41 open_flags = os.O_RDWR | os.O_TRUNC 

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

43 if o_nofollow is not None: 

44 open_flags |= o_nofollow 

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

46 open_flags |= os.O_CREAT 

47 fd = os.open(self.lock_file, open_flags, self._context.mode) 

48 with suppress(PermissionError): # This locked is not owned by this UID 

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

50 try: 

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

52 except OSError as exception: 

53 os.close(fd) 

54 if exception.errno == ENOSYS: # NotImplemented error 

55 msg = "FileSystem does not appear to support flock; use SoftFileLock instead" 

56 raise NotImplementedError(msg) from exception 

57 else: 

58 self._context.lock_file_fd = fd 

59 

60 def _release(self) -> None: 

61 # Do not remove the lockfile: 

62 # https://github.com/tox-dev/py-filelock/issues/31 

63 # https://stackoverflow.com/questions/17708885/flock-removing-locked-file-without-race-condition 

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

65 self._context.lock_file_fd = None 

66 fcntl.flock(fd, fcntl.LOCK_UN) 

67 os.close(fd) 

68 

69 

70__all__ = [ 

71 "UnixFileLock", 

72 "has_fcntl", 

73]