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]