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

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

41 statements  

1from __future__ import annotations 

2 

3import os 

4import sys 

5from contextlib import suppress 

6from errno import EACCES 

7from pathlib import Path 

8from typing import cast 

9 

10from ._api import BaseFileLock 

11from ._util import ensure_directory_exists, raise_on_not_writable_file 

12 

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

14 import msvcrt 

15 

16 class WindowsFileLock(BaseFileLock): 

17 """Uses the :func:`msvcrt.locking` function to hard lock the lock file on Windows systems.""" 

18 

19 def _acquire(self) -> None: 

20 raise_on_not_writable_file(self.lock_file) 

21 ensure_directory_exists(self.lock_file) 

22 flags = ( 

23 os.O_RDWR # open for read and write 

24 | os.O_CREAT # create file if not exists 

25 | os.O_TRUNC # truncate file if not empty 

26 ) 

27 try: 

28 fd = os.open(self.lock_file, flags, self._context.mode) 

29 except OSError as exception: 

30 if exception.errno != EACCES: # has no access to this lock 

31 raise 

32 else: 

33 try: 

34 msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) 

35 except OSError as exception: 

36 os.close(fd) # close file first 

37 if exception.errno != EACCES: # file is already locked 

38 raise 

39 else: 

40 self._context.lock_file_fd = fd 

41 

42 def _release(self) -> None: 

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

44 self._context.lock_file_fd = None 

45 msvcrt.locking(fd, msvcrt.LK_UNLCK, 1) 

46 os.close(fd) 

47 

48 with suppress(OSError): # Probably another instance of the application hat acquired the file lock. 

49 Path(self.lock_file).unlink() 

50 

51else: # pragma: win32 no cover 

52 

53 class WindowsFileLock(BaseFileLock): 

54 """Uses the :func:`msvcrt.locking` function to hard lock the lock file on Windows systems.""" 

55 

56 def _acquire(self) -> None: 

57 raise NotImplementedError 

58 

59 def _release(self) -> None: 

60 raise NotImplementedError 

61 

62 

63__all__ = [ 

64 "WindowsFileLock", 

65]