| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849 |
- from __future__ import annotations
- import os
- import sys
- from contextlib import suppress
- from errno import EACCES, EEXIST
- from pathlib import Path
- from ._api import BaseFileLock
- from ._util import ensure_directory_exists, raise_on_not_writable_file
- class SoftFileLock(BaseFileLock):
- """Simply watches the existence of the lock file."""
- def _acquire(self) -> None:
- raise_on_not_writable_file(self.lock_file)
- ensure_directory_exists(self.lock_file)
- flags = (
- os.O_WRONLY # open for writing only
- | os.O_CREAT
- | os.O_EXCL # together with above raise EEXIST if the file specified by filename exists
- | os.O_TRUNC # truncate the file to zero byte
- )
- o_nofollow = getattr(os, "O_NOFOLLOW", None)
- if o_nofollow is not None:
- flags |= o_nofollow
- try:
- file_handler = os.open(self.lock_file, flags, self._context.mode)
- except OSError as exception: # re-raise unless expected exception
- if not (
- exception.errno == EEXIST # lock already exist
- or (exception.errno == EACCES and sys.platform == "win32") # has no access to this lock
- ): # pragma: win32 no cover
- raise
- else:
- self._context.lock_file_fd = file_handler
- def _release(self) -> None:
- assert self._context.lock_file_fd is not None # noqa: S101
- os.close(self._context.lock_file_fd) # the lock file is definitely not None
- self._context.lock_file_fd = None
- with suppress(OSError): # the file is already deleted and that's what we want
- Path(self.lock_file).unlink()
- __all__ = [
- "SoftFileLock",
- ]
|
