Aluode/PerceptionLabPortable
0
1from __future__ import annotations2 3import os4import sys5from contextlib import suppress6from errno import EACCES, EEXIST7from pathlib import Path8 9from ._api import BaseFileLock10from ._util import ensure_directory_exists, raise_on_not_writable_file11 12 13class SoftFileLock(BaseFileLock):14 """Simply watches the existence of the lock file."""15 16 def _acquire(self) -> None:17 raise_on_not_writable_file(self.lock_file)18 ensure_directory_exists(self.lock_file)19 # first check for exists and read-only mode as the open will mask this case as EEXIST20 flags = (21 os.O_WRONLY # open for writing only22 | os.O_CREAT23 | os.O_EXCL # together with above raise EEXIST if the file specified by filename exists24 | os.O_TRUNC # truncate the file to zero byte25 )26 try:27 file_handler = os.open(self.lock_file, flags, self._context.mode)28 except OSError as exception: # re-raise unless expected exception29 if not (30 exception.errno == EEXIST # lock already exist31 or (exception.errno == EACCES and sys.platform == "win32") # has no access to this lock32 ): # pragma: win32 no cover33 raise34 else:35 self._context.lock_file_fd = file_handler36 37 def _release(self) -> None:38 assert self._context.lock_file_fd is not None # noqa: S10139 os.close(self._context.lock_file_fd) # the lock file is definitely not None40 self._context.lock_file_fd = None41 with suppress(OSError): # the file is already deleted and that's what we want42 Path(self.lock_file).unlink()43 44 45__all__ = [46 "SoftFileLock",47]48 