Installation

Install PM2 Python Library using pip:

pip install pm2

Or install from source:

git clone https://github.com/y4kupkaya/pm2.git
cd pm2
pip install -e .

Prerequisites: Make sure PM2 is installed globally:

npm install -g pm2

Quick Start

Here's a comprehensive example to get you started:

Synchronous Usage

from pm2 import PM2Manager

# Initialize PM2 manager
pm2 = PM2Manager()

# Start a new application
app = pm2.start_app("app.js", name="my-app", instances=4)
print(f"Started {app.name} with {app.instances} instances")

# List all processes
processes = pm2.list_processes()
for process in processes:
    print(f"{process.name}: {process.status.value} - {process.memory_usage_human}")

# Get specific process
my_app = pm2.get_process(name="my-app")
print(f"App uptime: {my_app.uptime_human}")

# Stop the process
pm2.stop_process(name="my-app")

Asynchronous Usage

import asyncio
from pm2 import PM2Manager

async def main():
    # Use async context manager
    async with PM2Manager() as pm2:
        # Start Python app
        app = await pm2.start_app_async("app.py", name="python-app")
        
        # Get process metrics
        process = await pm2.get_process_async(name="python-app")
        print(f"CPU: {process.metrics.cpu}%")
        print(f"Memory: {process.memory_usage_human}")
        
        # Get logs
        logs = await pm2.get_logs_async(name="python-app", lines=50)
        print("Recent logs:", logs)
        
        # Restart with zero downtime
        await pm2.reload_process_async(name="python-app")

asyncio.run(main())

Features

• 🚀 Full PM2 API coverage - start, stop, restart, delete, reload, etc.
• ⚡ Both sync and async support - Use with asyncio or traditional synchronous code
• 📊 Real-time process monitoring - CPU, memory metrics with human-readable formats
• 🔧 Advanced configuration management - Process environments, resource limits
• 📝 Comprehensive logging support - Log retrieval and management
• 🛡️ Robust error handling - Custom exceptions with detailed error information
• 🎯 Type hints - Full type annotation for better IDE support
• 🧪 Extensive test coverage - Production-ready reliability

PM2Manager

Main class for PM2 process management with comprehensive functionality.

PM2Manager.__init__()

Initialize a new PM2Manager instance with optional PM2 binary path validation.

Signature

PM2Manager(pm2_binary: str = "pm2", validate: bool = True)

Parameters

• pm2_binary (str): Path to PM2 executable. Defaults to "pm2".
• validate (bool): Whether to validate PM2 installation on init. Defaults to True.

Raises

• PM2ConnectionError: If PM2 is not installed or accessible.

Example

from pm2 import PM2Manager

# Default initialization
pm2 = PM2Manager()

# Custom PM2 path
pm2 = PM2Manager(pm2_binary="/usr/local/bin/pm2")

# Skip validation (faster init)
pm2 = PM2Manager(validate=False)

list_processes() / list_processes_async()

Get a list of all PM2 processes with detailed information.

Signature

def list_processes(self) -> List[PM2Process]
async def list_processes_async(self) -> List[PM2Process]

Returns

List[PM2Process]: List of PM2Process objects with complete process information.

Example

# Sync version
processes = pm2.list_processes()
for proc in processes:
    print(f"{proc.name}: {proc.status.value}")
    print(f"  PID: {proc.pid}, Uptime: {proc.uptime_human}")
    print(f"  CPU: {proc.metrics.cpu}%, Memory: {proc.memory_usage_human}")

# Async version
processes = await pm2.list_processes_async()
print(f"Total processes: {len(processes)}")

get_process() / get_process_async()

Get specific process by name, PID, or PM2 ID.

Signature

def get_process(self, name: Optional[str] = None, 
                pid: Optional[int] = None, 
                pm_id: Optional[int] = None) -> PM2Process

async def get_process_async(self, name: Optional[str] = None,
                           pid: Optional[int] = None,
                           pm_id: Optional[int] = None) -> PM2Process

Parameters

• name (str, optional): Process name.
• pid (int, optional): Process ID.
• pm_id (int, optional): PM2 internal ID.

Returns

PM2Process: Process object with complete information.

Raises

• PM2ProcessNotFoundError: If process is not found.
• PM2ValidationError: If no identifier is provided.

Example

# Get by name
process = pm2.get_process(name="my-app")

# Get by PID
process = pm2.get_process(pid=1234)

# Get by PM2 ID
process = pm2.get_process(pm_id=0)

# Check process status
if process.is_online:
    print(f"Process is running for {process.uptime_human}")
    print(f"Memory usage: {process.memory_usage_human}")
    print(f"Environment: {process.environment.variables}")

start_app() / start_app_async()

Start a new application with PM2 with comprehensive configuration options.

Signature

def start_app(self, script: str, name: Optional[str] = None, **kwargs) -> PM2Process

async def start_app_async(self, script: str, name: Optional[str] = None, **kwargs) -> PM2Process

Parameters

• script (str): Path to the script or application to run.
• name (str, optional): Process name. Auto-generated if not provided.
• **kwargs: Additional PM2 configuration options.

Common kwargs

• instances (int|str): Number of instances to start (cluster mode).
• exec_mode (str): Execution mode (fork/cluster).
• env (dict): Environment variables.
• args (list): Arguments to pass to the script.

Returns

PM2Process: The started process object.

Example

# Simple start
app = pm2.start_app("server.js", name="web-server")

# Advanced configuration
app = pm2.start_app(
    "api.py",
    name="api-server",
    instances=4,  # Cluster mode with 4 instances
    env={
        "NODE_ENV": "production",
        "PORT": "3000",
        "DATABASE_URL": "postgresql://..."
    },
    args=["--port", "3000", "--workers", "4"]
)

print(f"Started {app.name} with {app.instances} instances")
print(f"Process ID: {app.pm_id}, PID: {app.pid}")

stop_process() / stop_process_async()

Stop a running process gracefully.

Signature

def stop_process(self, name: Optional[str] = None,
                 pid: Optional[int] = None,
                 pm_id: Optional[int] = None) -> PM2Process

async def stop_process_async(self, name: Optional[str] = None,
                            pid: Optional[int] = None,
                            pm_id: Optional[int] = None) -> PM2Process

Returns

PM2Process: The stopped process object with updated status.

Example

# Stop by name
stopped_process = pm2.stop_process(name="my-app")
print(f"Process {stopped_process.name} is now {stopped_process.status.value}")

# Stop by PM2 ID
await pm2.stop_process_async(pm_id=0)

restart_process() / restart_process_async()

Restart a process (stop then start).

Signature

def restart_process(self, name: Optional[str] = None,
                    pid: Optional[int] = None,
                    pm_id: Optional[int] = None) -> PM2Process

Returns

PM2Process: The restarted process object.

Example

# Restart by name
restarted = pm2.restart_process(name="api-server")
print(f"Restart count: {restarted.restart_time}")

delete_process() / delete_process_async()

Delete a process permanently from PM2.

Signature

def delete_process(self, name: Optional[str] = None,
                   pid: Optional[int] = None,
                   pm_id: Optional[int] = None) -> bool

Returns

bool: True if deletion was successful.

Example

# Delete by name
success = pm2.delete_process(name="old-app")
if success:
    print("Process deleted successfully")

reload_process() / reload_process_async()

Reload a process with zero-downtime restart (cluster mode only).

Signature

def reload_process(self, name: Optional[str] = None,
                   pid: Optional[int] = None,
                   pm_id: Optional[int] = None) -> PM2Process

Returns

PM2Process: The reloaded process object.

Example

# Zero-downtime reload
reloaded = pm2.reload_process(name="web-app")
print(f"Reloaded without downtime: {reloaded.name}")

get_logs() / get_logs_async()

Retrieve process logs.

Signature

def get_logs(self, name: Optional[str] = None,
             pid: Optional[int] = None,
             pm_id: Optional[int] = None,
             lines: int = 100) -> str

Parameters

• lines (int): Number of log lines to retrieve. Defaults to 100.

Returns

str: Log content as string.

Example

# Get recent logs
logs = pm2.get_logs(name="api-server", lines=50)
print("Recent logs:")
print(logs)

# Get all available logs
all_logs = pm2.get_logs(name="api-server", lines=1000)

flush_logs() / flush_logs_async()

Clear process logs.

Signature

def flush_logs(self, name: Optional[str] = None,
               pid: Optional[int] = None,
               pm_id: Optional[int] = None) -> bool

Returns

bool: True if logs were flushed successfully.

Example

# Flush specific process logs
pm2.flush_logs(name="api-server")

# Flush all logs
pm2.flush_logs()

save_process_list() / save_process_list_async()

Save the current process list for later resurrection.

Signature

def save_process_list(self) -> bool

Returns

bool: True if save was successful.

Example

# Save current processes
pm2.save_process_list()
print("Process list saved for resurrection")

resurrect_processes() / resurrect_processes_async()

Restore previously saved processes.

Signature

def resurrect_processes(self) -> List[PM2Process]

Returns

List[PM2Process]: List of resurrected processes.

Example

# Restore saved processes
processes = pm2.resurrect_processes()
print(f"Resurrected {len(processes)} processes")

kill_daemon() / kill_daemon_async()

Kill the PM2 daemon and all processes.

Signature

def kill_daemon(self) -> bool

Returns

bool: True if daemon was killed successfully.

Example

# Kill PM2 daemon
pm2.kill_daemon()
print("PM2 daemon stopped")

PM2Process

Enhanced PM2 process representation with advanced features and metrics.

Key Properties

• name (str): Process name
• pid (int): Process ID
• pm_id (int): PM2 internal ID
• status (ProcessStatus): Current status
• metrics (ProcessMetrics): CPU and memory metrics
• uptime_human (str): Human-readable uptime
• memory_usage_human (str): Human-readable memory usage
• environment (ProcessEnvironment): Environment variables

Example

process = pm2.get_process(name="my-app")

print(f"Name: {process.name}")
print(f"Status: {process.status.value}")
print(f"PID: {process.pid}")
print(f"Uptime: {process.uptime_human}")
print(f"CPU: {process.metrics.cpu}%")
print(f"Memory: {process.memory_usage_human}")
print(f"Restart count: {process.restart_time}")

# Check if online
if process.is_online:
    print("Process is running")

# Get environment variables
db_url = process.get_environment_var("DATABASE_URL")
print(f"Database URL: {db_url}")

# Convert to dict/JSON
process_data = process.to_dict()
process_json = process.to_json(indent=2)

Async Context Manager

Use PM2Manager as an async context manager for better resource management.

Example

import asyncio
from pm2 import PM2Manager

async def manage_processes():
    async with PM2Manager() as pm2:
        # All async methods are available
        processes = await pm2.list_processes_async()
        
        for process in processes:
            if not process.is_online:
                print(f"Restarting {process.name}")
                await pm2.restart_process_async(name=process.name)
        
        # Context manager handles cleanup
        
asyncio.run(manage_processes())

Async Methods

All PM2Manager methods have async counterparts with _async suffix.

Available Async Methods

• list_processes_async()
• get_process_async()
• start_app_async()
• stop_process_async()
• restart_process_async()
• delete_process_async()
• reload_process_async()
• get_logs_async()
• flush_logs_async()
• save_process_list_async()
• resurrect_processes_async()
• kill_daemon_async()

Example

async def batch_operations():
    pm2 = PM2Manager()
    
    # Start multiple apps concurrently
    tasks = [
        pm2.start_app_async("app1.js", name="app1"),
        pm2.start_app_async("app2.py", name="app2"),
        pm2.start_app_async("app3.js", name="app3")
    ]
    
    started_apps = await asyncio.gather(*tasks)
    print(f"Started {len(started_apps)} applications")
    
    # Check status of all apps
    processes = await pm2.list_processes_async()
    for proc in processes:
        print(f"{proc.name}: {proc.status.value}")

ProcessMetrics

Process monitoring metrics with memory calculations.

Properties

• cpu (float): CPU usage percentage
• memory (int): Memory usage in bytes
• heap_used (int): Heap memory used
• heap_total (int): Total heap memory
• external (int): External memory
• rss (int): RSS memory
• memory_mb (float): Memory in MB (property)
• heap_used_mb (float): Heap used in MB (property)

Example

process = pm2.get_process(name="my-app")
metrics = process.metrics

print(f"CPU: {metrics.cpu}%")
print(f"Memory: {metrics.memory_mb:.1f} MB")
print(f"Heap Used: {metrics.heap_used_mb:.1f} MB")
print(f"RSS: {metrics.rss} bytes")

ProcessEnvironment

Process environment configuration management.

Properties

• variables (Dict[str, str]): Environment variables
• node_version (str): Node.js version
• python_version (str): Python version
• virtual_env (str): Virtual environment path
• working_directory (str): Working directory

Methods

• set_var(key, value): Set environment variable
• get_var(key, default): Get environment variable

Example

from pm2 import ProcessEnvironment

env = ProcessEnvironment()
env.set_var("NODE_ENV", "production")
env.set_var("PORT", "3000")

node_env = env.get_var("NODE_ENV", "development")
print(f"Environment: {node_env}")

ProcessConfiguration

Advanced process configuration for PM2 ecosystem format.

Key Properties

• name: Process name
• script: Script to execute
• instances: Number of instances
• exec_mode: Execution mode (fork/cluster)
• max_memory_restart: Memory limit for restart
• max_restarts: Maximum restart count
• autorestart: Auto restart on crash
• watch: Watch for file changes
• env: Environment configuration

Example

from pm2 import ProcessConfiguration, ProcessMode, ProcessEnvironment

# Create environment
env = ProcessEnvironment()
env.set_var("NODE_ENV", "production")

# Create configuration
config = ProcessConfiguration(
    name="advanced-app",
    script="server.js",
    instances=4,
    exec_mode=ProcessMode.CLUSTER,
    max_memory_restart="1G",
    max_restarts=5,
    autorestart=True,
    watch=["src", "config"],
    env=env
)

# Convert to PM2 ecosystem format
ecosystem_config = config.to_dict()
print(json.dumps(ecosystem_config, indent=2))

ProcessStatus

Enumeration for PM2 process statuses.

Values

• ONLINE: "online" - Process is running
• STOPPING: "stopping" - Process is stopping
• STOPPED: "stopped" - Process is stopped
• LAUNCHING: "launching" - Process is starting
• ERRORED: "errored" - Process has errors
• ONE_LAUNCH_STATUS: "one-launch-status" - One-time execution

ProcessMode

Enumeration for PM2 execution modes.

Values

• FORK: "fork" - Fork mode (single instance)
• CLUSTER: "cluster" - Cluster mode (multiple instances)

LogLevel

Enumeration for logging levels.

Values

• ERROR: "error"
• WARN: "warn"
• INFO: "info"
• VERBOSE: "verbose"
• DEBUG: "debug"
• SILENT: "silent"

PM2Error

Base exception for all PM2-related errors with detailed information.

Properties

• message (str): Error message
• details (dict): Additional error details
• timestamp (datetime): Error timestamp

Example

try:
    pm2.get_process(name="nonexistent")
except PM2Error as e:
    print(f"Error: {e.message}")
    print(f"Details: {e.details}")
    print(f"Time: {e.timestamp}")

PM2ConnectionError

Raised when PM2 daemon connection fails.

PM2CommandError

Raised when PM2 command execution fails.

Additional Properties

• command (list): The failed command
• exit_code (int): Command exit code

PM2ProcessError

Base class for process-related errors.

PM2ProcessNotFoundError

Raised when a requested process is not found.

Additional Properties

• identifier (str): The identifier that was searched for
• identifier_type (str): Type of identifier (name, pid, pm_id)

PM2ProcessAlreadyExistsError

Raised when trying to start a process that already exists.

PM2ProcessInvalidStateError

Raised when process is in invalid state for requested operation.

PM2ConfigurationError

Raised when PM2 configuration is invalid.

PM2ValidationError

Raised when input validation fails.

PathIsFolderError

Raised when a folder path is provided instead of a file path.

PM2CommandExecutor

Handles PM2 command execution with proper error handling and validation.

Methods

• execute(args, timeout): Execute PM2 command synchronously
• execute_async(args, timeout): Execute PM2 command asynchronously

Example

from pm2 import PM2CommandExecutor

executor = PM2CommandExecutor()

# Execute command
result = executor.execute(["list"])
print(result["stdout"])

# Async execution
result = await executor.execute_async(["status"])
print(result["returncode"])