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

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 except ImportError: 

30 pass 

31 else: 

32 has_fcntl = True 

33 

34 class UnixFileLock(BaseFileLock): 

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

36 

37 def _acquire(self) -> None: 

38 ensure_directory_exists(self.lock_file) 

39 open_flags = os.O_RDWR | os.O_TRUNC 

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

41 open_flags |= os.O_CREAT 

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

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

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

45 try: 

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

47 except OSError as exception: 

48 os.close(fd) 

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

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

51 raise NotImplementedError(msg) from exception 

52 else: 

53 self._context.lock_file_fd = fd 

54 

55 def _release(self) -> None: 

56 # Do not remove the lockfile: 

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

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

59 fd = cast(int, self._context.lock_file_fd) 

60 self._context.lock_file_fd = None 

61 fcntl.flock(fd, fcntl.LOCK_UN) 

62 os.close(fd) 

63 

64 

65__all__ = [ 

66 "UnixFileLock", 

67 "has_fcntl", 

68]