Utilities for counting running instances of an application.

CountHelper 

CountHelper(*, lock_dir: str | None = None)

Class to count running instances of an application by process and lock files.

This class provides methods to count how many instances of an application are currently running either via checking process command lines or scanning for lock files in a specified directory.

Parameters:
  • lock_dir (str | None, default: None ) –

    Optional path to the directory where lock files are stored. If not provided, uses the system's temporary directory.
Example 
print(CountHelper(lock_dir="/tmp/my_app_locks").lock_dir)
#> /tmp/my_app_locks

process_count classmethod 

process_count(identifier: str) -> int

Count the number of running processes that match the given identifier.

Iterates over all running processes and checks if the identifier appears in their command-line arguments.

Parameters:
  • identifier (str) –

    String to search for in process command lines.
Returns:
  • int –

    Number of matching processes.
Raises:
  • Exception –

    Any exception raised by 'psutil.process_iter'.
Example 
print(CountHelper.process_count("my_app"))
#> 2

pid_file_count 

pid_file_count(identifier: str) -> int

Count the number of lock files in the lock directory by identifier.

Lock files are expected to have names containing .lock. and the identifier.

Parameters:
  • identifier (str) –

    String to search for in lock file names.
Returns:
  • int –

    Number of matching lock files.
Example 
helper = CountHelper(lock_dir="/tmp/locks")
print(helper.pid_file_count("my_app"))
#> 1

acquire_pid_file 

acquire_pid_file(filename: str) -> tuple[Path, int]

Create and open a lock file with the given filename, and lock it.

The file is created with write access and locked using os.lockf with F_LOCK.

Parameters:
  • filename (str) –

    Name of the lock file to create.
Returns:
  • tuple[Path, int] –

    A tuple containing the path to the file and its file descriptor.
Example 
helper = CountHelper(lock_dir="/tmp/locks")
path, fd = helper.acquire_pid_file("my_app.lock.123456")
print(path)
#> /tmp/locks/my_app.lock.123456

is_released_pid_file 

is_released_pid_file(filename: str) -> bool

Check whether a lock file is available (not locked by another process).

Tries to test the lock using os.lockf with F_TEST. Returns True if the file is available, False if it is locked.

Parameters:
  • filename (str) –

    Name of the lock file to check.
Returns:
  • bool –

    True if the file is available, False otherwise.
Example 
helper = CountHelper(lock_dir="/tmp/locks")
print(helper.is_released_pid_file("my_app.lock.123456"))
#> True

get_pid_file 

get_pid_file(identifier: str) -> Path | None

Get the first lock file in the lock directory that matches the identifier.

Parameters:
  • identifier (str) –

    String to search for in lock file names.
Returns:
  • Path | None –

    Path to the first matching lock file, or None if no match found.
Example 
helper = CountHelper(lock_dir="/tmp/locks")
print(str(helper.get_pid_file("my_app")))
#> /tmp/locks/my_app.lock.123456