_unix.py 2.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970
  1. from __future__ import annotations
  2. import os
  3. import sys
  4. from contextlib import suppress
  5. from errno import ENOSYS
  6. from pathlib import Path
  7. from typing import cast
  8. from ._api import BaseFileLock
  9. from ._util import ensure_directory_exists
  10. #: a flag to indicate if the fcntl API is available
  11. has_fcntl = False
  12. if sys.platform == "win32": # pragma: win32 cover
  13. class UnixFileLock(BaseFileLock):
  14. """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems."""
  15. def _acquire(self) -> None:
  16. raise NotImplementedError
  17. def _release(self) -> None:
  18. raise NotImplementedError
  19. else: # pragma: win32 no cover
  20. try:
  21. import fcntl
  22. _ = (fcntl.flock, fcntl.LOCK_EX, fcntl.LOCK_NB, fcntl.LOCK_UN)
  23. except (ImportError, AttributeError):
  24. pass
  25. else:
  26. has_fcntl = True
  27. class UnixFileLock(BaseFileLock):
  28. """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems."""
  29. def _acquire(self) -> None:
  30. ensure_directory_exists(self.lock_file)
  31. open_flags = os.O_RDWR | os.O_TRUNC | os.O_NOFOLLOW
  32. if not Path(self.lock_file).exists():
  33. open_flags |= os.O_CREAT
  34. fd = os.open(self.lock_file, open_flags, self._context.mode)
  35. with suppress(PermissionError): # This locked is not owned by this UID
  36. os.fchmod(fd, self._context.mode)
  37. try:
  38. fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
  39. except OSError as exception:
  40. os.close(fd)
  41. if exception.errno == ENOSYS: # NotImplemented error
  42. msg = "FileSystem does not appear to support flock; use SoftFileLock instead"
  43. raise NotImplementedError(msg) from exception
  44. else:
  45. self._context.lock_file_fd = fd
  46. def _release(self) -> None:
  47. # Do not remove the lockfile:
  48. # https://github.com/tox-dev/py-filelock/issues/31
  49. # https://stackoverflow.com/questions/17708885/flock-removing-locked-file-without-race-condition
  50. fd = cast("int", self._context.lock_file_fd)
  51. self._context.lock_file_fd = None
  52. fcntl.flock(fd, fcntl.LOCK_UN)
  53. os.close(fd)
  54. __all__ = [
  55. "UnixFileLock",
  56. "has_fcntl",
  57. ]