From 655a3914268cf5bc650a24f18b6754f3028510a1 Mon Sep 17 00:00:00 2001 From: lidong Date: Fri, 14 Jun 2024 01:34:15 +0800 Subject: [PATCH] add `is_running`, `lock_pid_file` --- CHANGELOG.md | 4 ++ README.md | 6 +++ doc.md | 45 +++++++++++++++++++++++ morebuiltins/utils.py | 85 +++++++++++++++++++++++++++++++++++++++++++ test.py | 1 + 5 files changed, 141 insertions(+) create mode 100644 CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..5b9b980 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,4 @@ +### v0.0.6 +1. add `xor_encode_decode` +2. add `is_running`, `lock_pid_file` +3. diff --git a/README.md b/README.md index a92d2b9..23442b8 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,8 @@ print(morebuiltins.__file__) ## Doc +[Changelog](https://github.com/ClericPy/morebuiltins/blob/master/doc.md) comes after v0.0.6(2024-06-14) + --- Module Docs - https://github.com/ClericPy/morebuiltins/blob/master/doc.md @@ -79,6 +81,10 @@ Module Docs - https://github.com/ClericPy/morebuiltins/blob/master/doc.md 1.20 `xor_encode_decode` - +1.21 `is_running` - Check if the given process ID is running. + +1.22 `set_pid_file` - Attempt to lock a PID file to ensure only one instance is running, like a singleton-lock. + ## 2. morebuiltins.functools diff --git a/doc.md b/doc.md index b547889..e927868 100644 --- a/doc.md +++ b/doc.md @@ -424,6 +424,51 @@ True +--- + + + 1.21 `is_running` - Check if the given process ID is running. + + Parameters: + pid -- The process ID to check. + + Returns: + True if the process is running; False otherwise. + + Examples: + >>> is_running(os.getpid() * 10) # Assume process is not running + False + >>> is_running(os.getpid()) # Check if the current process is running + True + >>> is_running("not_a_pid") # Invalid PID input should be handled and return False + False + + + +--- + + + 1.22 `set_pid_file` - Attempt to lock a PID file to ensure only one instance is running, like a singleton-lock. + + Args: + - path (Union[str, Path]): The path to the PID file. + - raise_error (bool): If True, raise an exception when the PID file exists and + corresponds to a running process. Defaults to True. + + Returns: + - bool: True if the PID file is successfully locked, otherwise, based on + `raise_error`, either raises an exception or returns False indicating + the lock could not be acquired. + + Examples: + + >>> set_pid_file("myapp.pid") # Assuming this is the first run, should succeed + True + >>> set_pid_file("myapp.pid") # Simulating second instance trying to start, should raise error if raise_error=True + False + >>> os.unlink("myapp.pid") + + --- ====================== diff --git a/morebuiltins/utils.py b/morebuiltins/utils.py index 255e2a8..d4a78c8 100644 --- a/morebuiltins/utils.py +++ b/morebuiltins/utils.py @@ -3,13 +3,16 @@ import gzip import hashlib import json +import os import re +import sys import traceback from collections import UserDict from enum import IntEnum from functools import wraps from itertools import groupby, islice from os.path import basename +from pathlib import Path from time import gmtime, mktime, strftime, strptime, time, timezone from typing import ( Any, @@ -46,6 +49,8 @@ "Trie", "GuessExt", "xor_encode_decode", + "is_running", + "set_pid_file", ] @@ -837,6 +842,86 @@ def xor_encode_decode(data, key): return bytes([b ^ k for b, k in zip(data, extended_key)]) +def is_running(pid): + """Check if the given process ID is running. + + Parameters: + pid -- The process ID to check. + + Returns: + True if the process is running; False otherwise. + + Examples: + >>> is_running(os.getpid() * 10) # Assume process is not running + False + >>> is_running(os.getpid()) # Check if the current process is running + True + >>> is_running("not_a_pid") # Invalid PID input should be handled and return False + False + + """ + try: + pid = int(pid) + except ValueError: + return False + if sys.platform == "win32": + with os.popen('tasklist /fo csv /fi "pid eq %s"' % int(pid)) as f: + f.readline() + text = f.readline() + return bool(text) + else: + try: + os.kill(int(pid), 0) + return True + except OSError: + return False + except SystemError: + # maybe windows? + return True + + +def set_pid_file(path: Union[str, Path], raise_error=False): + """Attempt to lock a PID file to ensure only one instance is running, like a singleton-lock. + + Args: + - path (Union[str, Path]): The path to the PID file. + - raise_error (bool): If True, raise an exception when the PID file exists and + corresponds to a running process. Defaults to True. + + Returns: + - bool: True if the PID file is successfully locked, otherwise, based on + `raise_error`, either raises an exception or returns False indicating + the lock could not be acquired. + + Examples: + + >>> set_pid_file("myapp.pid") # Assuming this is the first run, should succeed + True + >>> set_pid_file("myapp.pid") # Simulating second instance trying to start, should raise error if raise_error=True + False + >>> os.unlink("myapp.pid") + """ + if isinstance(path, str): + path = Path(path) + running = False + if path.is_file(): + try: + old_pid = int(path.read_text().strip()) + running = is_running(old_pid) + except ValueError: + # non-int pid, NaN + pass + + if running: + if raise_error: + raise RuntimeError(f"{path.as_posix()} is locked by {old_pid}") + else: + return False + else: + path.write_text(f"{os.getpid()}") + return True + + if __name__ == "__main__": __name__ = "morebuiltins.utils" import doctest diff --git a/test.py b/test.py index 9ff97ff..7a11b89 100644 --- a/test.py +++ b/test.py @@ -47,6 +47,7 @@ def test_all(): module.__name__, flush=True, ) + time.sleep(1) print("all test ok", flush=True)