common.py

Overview

common.py is a utility module providing a safe and convenient way to run shell commands asynchronously within an asyncio event loop environment. It defines a single asynchronous function async_run_command which executes an external command with a specified timeout, capturing its output (stdout and stderr) and exit code. The function handles common errors such as timeouts and unexpected process termination reliably.

This module is useful in applications that require non-blocking execution of shell commands, such as network automation, system orchestration, or any async Python program that interacts with external processes without stalling the main event loop.

Detailed Description

Function: async_run_command

async def async_run_command(*args, timeout: float = 5) -> Tuple[int, str, str]:

Purpose

Runs an external command asynchronously, waits for it to complete or the timeout to expire, and returns the process's exit code along with its stdout and stderr outputs decoded as strings.

Parameters

• *args:
  Variable length argument list representing the command and its arguments to execute. For example, async_run_command("ls", "-l") runs the command ls -l.

• timeout (float, optional, default=5):
  Maximum number of seconds to wait for the command to complete. If the command runs longer, it will be forcibly terminated.

Returns

• Tuple[int, str, str]:
  A tuple containing:

  • int: The process exit code (proc.returncode).

  • str: The decoded standard output (stdout) of the process.

  • str: The decoded standard error (stderr) of the process.

Raises

• RuntimeError:

  • If the process finishes but its return code is None, which is an abnormal state.

  • If the command exceeds the specified timeout.

• Any other exception raised during command execution is propagated after ensuring the subprocess is terminated.

Usage Example

import asyncio
from common import async_run_command

async def main():
    try:
        retcode, out, err = await async_run_command("ls", "-l", timeout=3)
        print(f"Exit code: {retcode}")
        print(f"Output:\n{out}")
        if err:
            print(f"Errors:\n{err}")
    except RuntimeError as e:
        print(f"Command failed: {e}")

asyncio.run(main())

Implementation Details

• The function uses asyncio.create_subprocess_exec to spawn the subprocess asynchronously and capture both stdout and stderr using pipes.

• It employs asyncio.wait_for to enforce the timeout constraint.

• If the timeout expires, the process is killed and awaited to ensure proper cleanup.

• Output streams from the process are decoded using default UTF-8 encoding.

• The subprocess return code is checked to confirm process completion; if the return code is None, it raises a RuntimeError to indicate an unexpected state.

• Any exceptions during execution cause subprocess termination to avoid orphaned processes.

Interaction with Other System Components

• This utility function can be used by any part of the application that requires running shell commands asynchronously.

• It provides a reliable foundation for higher-level modules that perform system calls, network commands, or external tool integrations.

• Because it is a low-level utility, it does not depend on other modules in the codebase but can be imported into components needing async command execution.

• The function’s design allows it to integrate seamlessly into event-driven architectures that use Python’s asyncio.

Visual Diagram

The following flowchart illustrates the workflow of the async_run_command function:

flowchart TD
    A[Start async_run_command(*args, timeout)] --> B[Create subprocess with asyncio.create_subprocess_exec]
    B --> C[Wait for process completion with asyncio.wait_for(timeout)]
    C -->|Success| D{Is proc.returncode None?}
    D -->|Yes| E[Raise RuntimeError("Process finished but returncode is None")]
    D -->|No| F[Decode stdout and stderr]
    F --> G[Return (returncode, stdout, stderr)]
    C -->|TimeoutError| H[Kill process]
    H --> I[Await process termination]
    I --> J[Raise RuntimeError("Command timed out")]
    C -->|Other Exception| K[Kill process]
    K --> L[Await process termination]
    L --> M[Re-raise exception]

Summary

common.py is a concise but critical utility that ensures asynchronous execution of shell commands with proper timeout handling and output collection. Its robust error handling and non-blocking design make it suitable for integration into complex asynchronous Python applications that interact with the system shell or external binaries.