import signal
import sys
from typing import Optional
+import warnings
import config
import datetime_utils
return True
except OSError:
pass
- logger.warning(f'Could not acquire {self.lockfile}.')
+ msg = f'Could not acquire {self.lockfile}.'
+ logger.warning(msg)
return False
def acquire_with_retries(
duration = ts - self.locktime
if duration >= config.config['lockfile_held_duration_warning_threshold_sec']:
str_duration = datetime_utils.describe_duration_briefly(duration)
- logger.warning(f'Held {self.lockfile} for {str_duration}')
+ msg = f'Held {self.lockfile} for {str_duration}'
+ logger.warning(msg)
+ warnings.warn(msg, stacklevel=2)
self.release()
def __del__(self):
try:
os.kill(contents.pid, 0)
except OSError:
- logger.warning(f'Lockfile {self.lockfile}\'s pid ({contents.pid}) is stale; ' +
- 'force acquiring')
+ msg = f'Lockfile {self.lockfile}\'s pid ({contents.pid}) is stale; force acquiring'
+ logger.warning(msg)
self.release()
# Has the lock expiration expired?
if contents.expiration_timestamp is not None:
now = datetime.datetime.now().timestamp()
if now > contents.expiration_datetime:
- logger.warning(f'Lockfile {self.lockfile} expiration time has passed; ' +
- 'force acquiring')
+ msg = f'Lockfile {self.lockfile} expiration time has passed; force acquiring'
+ logger.warning(msg)
self.release()
except Exception:
pass