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

1from __future__ import annotations 

2 

3import os 

4import sys 

5from contextlib import suppress 

6from errno import ENOSYS 

7from typing import cast 

8 

9from ._api import BaseFileLock 

10from ._util import ensure_directory_exists 

11 

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

13has_fcntl = False 

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

15 

16 class UnixFileLock(BaseFileLock): 

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

18 

19 def _acquire(self) -> None: 

20 raise NotImplementedError 

21 

22 def _release(self) -> None: 

23 raise NotImplementedError 

24 

25else: # pragma: win32 no cover 

26 try: 

27 import fcntl 

28 except ImportError: 

29 pass 

30 else: 

31 has_fcntl = True 

32 

33 class UnixFileLock(BaseFileLock): 

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

35 

36 def _acquire(self) -> None: 

37 ensure_directory_exists(self.lock_file) 

38 open_flags = os.O_RDWR | os.O_CREAT | os.O_TRUNC 

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

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

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

42 try: 

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

44 except OSError as exception: 

45 os.close(fd) 

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

47 msg = "FileSystem does not appear to support flock; user SoftFileLock instead" 

48 raise NotImplementedError(msg) from exception 

49 else: 

50 self._context.lock_file_fd = fd 

51 

52 def _release(self) -> None: 

53 # Do not remove the lockfile: 

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

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

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

57 self._context.lock_file_fd = None 

58 fcntl.flock(fd, fcntl.LOCK_UN) 

59 os.close(fd) 

60 

61 

62__all__ = [ 

63 "has_fcntl", 

64 "UnixFileLock", 

65]