*,
do_signal_cleanup: bool = True,
expiration_timestamp: Optional[float] = None,
+ override_command: Optional[str] = None,
) -> None:
self.is_locked = False
self.lockfile = lockfile_path
+ self.override_command = override_command
if do_signal_cleanup:
signal.signal(signal.SIGINT, self._signal)
signal.signal(signal.SIGTERM, self._signal)
self.release()
def _get_lockfile_contents(self) -> str:
+ if self.override_command:
+ cmd = self.override_command
+ else:
+ cmd = ' '.join(sys.argv)
+ print(cmd)
contents = LockFileContents(
pid = os.getpid(),
- commandline = ' '.join(sys.argv),
- expiration_timestamp = self.expiration_timestamp
+ commandline = cmd,
+ expiration_timestamp = self.expiration_timestamp,
)
return json.dumps(contents.__dict__)