"""File-based locking helper."""
+from __future__ import annotations
+import contextlib
import datetime
import json
import logging
import sys
import warnings
from dataclasses import dataclass
-from typing import Optional
+from typing import Literal, Optional
import config
import datetime_utils
expiration_timestamp: Optional[float]
-class LockFile(object):
+class LockFile(contextlib.AbstractContextManager):
"""A file locking mechanism that has context-manager support so you
can use it in a with statement. e.g.
logger.warning(msg)
raise LockFileException(msg)
- def __exit__(self, _, value, traceback):
+ def __exit__(self, _, value, traceback) -> Literal[False]:
if self.locktime:
ts = datetime.datetime.now().timestamp()
duration = ts - self.locktime
if duration >= config.config['lockfile_held_duration_warning_threshold_sec']:
- str_duration = datetime_utils.describe_duration_briefly(duration)
+ # Note: describe duration briefly only does 1s granularity...
+ str_duration = datetime_utils.describe_duration_briefly(int(duration))
msg = f'Held {self.lockfile} for {str_duration}'
logger.warning(msg)
warnings.warn(msg, stacklevel=2)
self.release()
+ return False
def __del__(self):
if self.is_locked:
try:
os.kill(contents.pid, 0)
except OSError:
- msg = f'Lockfile {self.lockfile}\'s pid ({contents.pid}) is stale; force acquiring'
- logger.warning(msg)
+ logger.warning(
+ 'Lockfile %s\'s pid (%d) is stale; force acquiring...',
+ self.lockfile,
+ contents.pid,
+ )
self.release()
# Has the lock expiration expired?
if contents.expiration_timestamp is not None:
now = datetime.datetime.now().timestamp()
if now > contents.expiration_timestamp:
- msg = f'Lockfile {self.lockfile} expiration time has passed; force acquiring'
- logger.warning(msg)
+ logger.warning(
+ 'Lockfile %s\'s expiration time has passed; force acquiring',
+ self.lockfile,
+ )
self.release()
except Exception:
- pass
+ pass # If the lockfile doesn't exist or disappears, good.