mirror of
https://github.com/alexgetmancom/miband-bot.git
synced 2026-09-05 18:16:18 +03:00
Alex
32 lines
937 B
Python
32 lines
937 B
Python
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import os
|
|
import time
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
|
|
class LockUnavailable(RuntimeError):
|
|
"""Raised when another process already owns the sync lock."""
|
|
|
|
|
|
@contextmanager
|
|
def exclusive_file_lock(path: Path) -> Iterator[None]:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a+", encoding="utf-8") as lock_file:
|
|
try:
|
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError as exc:
|
|
raise LockUnavailable(f"Lock is already held: {path}") from exc
|
|
|
|
try:
|
|
lock_file.seek(0)
|
|
lock_file.truncate()
|
|
lock_file.write(f"pid={os.getpid()} time={int(time.time())}\n")
|
|
lock_file.flush()
|
|
yield
|
|
finally:
|
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
|