mirror of
https://github.com/Chia-Network/chia-blockchain.git
synced 2026-08-29 02:24:35 -05:00
Matt HauffandGitHub
40db4635a8
* Enable PEP604 Ruff rules * Fix harcoded signature in test * Hack CLVMStreamable test with note to fast follow
49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
# Package: utils
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from types import TracebackType
|
|
|
|
from filelock import BaseFileLock, FileLock, Timeout
|
|
from typing_extensions import final
|
|
|
|
|
|
class LockfileError(Exception):
|
|
pass
|
|
|
|
|
|
@final
|
|
@dataclass(frozen=True)
|
|
class Lockfile:
|
|
_lock: BaseFileLock
|
|
timeout: float
|
|
poll_interval: float
|
|
|
|
@classmethod
|
|
def create(cls, path: Path, timeout: float = -1, poll_interval: float = 0.05) -> Lockfile:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
return cls(_lock=FileLock(path.with_name(path.name + ".lock")), timeout=timeout, poll_interval=poll_interval)
|
|
|
|
def __enter__(self) -> Lockfile:
|
|
self.acquire(timeout=self.timeout, poll_interval=self.poll_interval)
|
|
return self
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: type[BaseException] | None,
|
|
exc: BaseException | None,
|
|
traceback: TracebackType | None,
|
|
) -> None:
|
|
self.release()
|
|
|
|
def acquire(self, timeout: float, poll_interval: float) -> None:
|
|
try:
|
|
self._lock.acquire(timeout=timeout, poll_interval=poll_interval)
|
|
except Timeout as e:
|
|
raise LockfileError(e) from e
|
|
|
|
def release(self) -> None:
|
|
self._lock.release()
|