Shell Tasks

Shell Tasks

Shell Tasks provide a robust and secure mechanism for executing external shell commands and scripts directly from within an application. This capability is essential for integrating with system utilities, automating operational workflows, and extending application functionality beyond its native scope.

Core Capabilities

Shell Tasks offer a comprehensive set of features to manage external process execution effectively.

Synchronous Command Execution

Shell Tasks execute commands synchronously, blocking the calling thread until the command completes. This is suitable for operations where the application must wait for the result before proceeding.

from shell_tasks import ShellTaskExecutor

executor = ShellTaskExecutor()
result = executor.execute_command("ls -l /tmp")
print(f"Command output:\n{result.stdout}")
if result.return_code != 0:
    print(f"Error: {result.stderr}")

The execute_command method returns a CommandResult object containing stdout, stderr, and return_code.

Asynchronous Command Execution

For long-running operations or scenarios requiring concurrent execution, Shell Tasks support asynchronous command execution. This allows the application to initiate a command and continue processing without waiting for its completion, later retrieving the result.

import asyncio
from shell_tasks import ShellTaskExecutor

async def run_async_task():
    executor = ShellTaskExecutor()
    task = await executor.run_async("sleep 5 && echo 'Done sleeping'")
    print("Command started, continuing with other tasks...")
    # Perform other operations
    await asyncio.sleep(1)
    print("One second passed...")
    result = await task.wait_for_completion()
    print(f"Async command output: {result.stdout}")

# asyncio.run(run_async_task())

The run_async method returns a ShellTask object, which provides methods like wait_for_completion to retrieve the CommandResult when the process finishes.

Input/Output Handling

Shell Tasks facilitate comprehensive control over standard input (stdin), standard output (stdout), and standard error (stderr) streams.

• Capturing Output: By default, stdout and stderr are captured and returned as strings in the CommandResult.
• Providing Input: Applications can pipe data to a command's stdin.

from shell_tasks import ShellTaskExecutor

executor = ShellTaskExecutor()
# Providing input via stdin
result = executor.execute_command("grep 'hello'", input_data="hello world\ngoodbye world")
print(f"Grep output: {result.stdout}") # Expected: "hello world"

Error Management

Shell Tasks provide robust error handling by capturing the command's exit code and stderr. A non-zero return_code in the CommandResult indicates an error.

from shell_tasks import ShellTaskExecutor, CommandExecutionError

executor = ShellTaskExecutor()
try:
    # Attempt to execute a non-existent command
    result = executor.execute_command("non_existent_command")
    if result.return_code != 0:
        print(f"Command failed with exit code {result.return_code}: {result.stderr}")
except CommandExecutionError as e:
    print(f"An execution error occurred: {e}")

For critical failures, such as the inability to spawn the process, Shell Tasks raise a CommandExecutionError.

Environment and Working Directory Control

Applications can specify the environment variables and the working directory for the executed command. This ensures commands run in the correct context, isolated from the parent process's environment if necessary.

from shell_tasks import ShellTaskExecutor

executor = ShellTaskExecutor()
result = executor.execute_command(
    "env | grep MY_VAR",
    env={"MY_VAR": "custom_value", "PATH": "/usr/bin:/bin"},
    cwd="/tmp"
)
print(f"Environment output: {result.stdout}") # Expected: "MY_VAR=custom_value"

Common Use Cases

Shell Tasks are invaluable for a variety of integration and automation scenarios.

• System Automation: Perform file system operations (e.g., cp, mv, rm), manage processes (e.g., kill, ps), or query system information (e.g., df, du).
• Tool Integration: Interact with external command-line tools like git, docker, kubectl, ffmpeg, or custom scripts. This allows applications to leverage existing CLI ecosystems without reimplementing their functionality.
• Build and Deployment Workflows: Orchestrate steps in a CI/CD pipeline, such as running build scripts, deploying artifacts, or executing database migrations.
• Data Processing: Execute data transformation scripts written in other languages (e.g., awk, sed, custom Python/Perl scripts) as part of a larger data pipeline.

Implementation Details and Examples

Executing a Simple Command

The most straightforward use involves running a command and getting its output.

from shell_tasks import ShellTaskExecutor

executor = ShellTaskExecutor()
result = executor.execute_command("echo 'Hello from shell!'")
print(f"Stdout: {result.stdout.strip()}")
print(f"Return Code: {result.return_code}")

Capturing Output

By default, both stdout and stderr are captured.

from shell_tasks import ShellTaskExecutor

executor = ShellTaskExecutor()
result = executor.execute_command("ls /non_existent_path")
print(f"Stdout (empty usually): '{result.stdout.strip()}'")
print(f"Stderr: {result.stderr.strip()}")
print(f"Return Code: {result.return_code}") # Will be non-zero

Handling Errors

Always check the return_code to determine command success.

from shell_tasks import ShellTaskExecutor

executor = ShellTaskExecutor()
result = executor.execute_command("false") # 'false' command exits with non-zero
if result.return_code != 0:
    print(f"Command failed with exit code {result.return_code}")
    print(f"Error details: {result.stderr.strip()}")
else:
    print("Command succeeded.")

Providing Input

Pass a string to the input_data parameter.

from shell_tasks import ShellTaskExecutor

executor = ShellTaskExecutor()
input_text = "line1\nline2\nline3"
result = executor.execute_command("wc -l", input_data=input_text)
print(f"Line count: {result.stdout.strip()}") # Expected: "3"

Running Commands Asynchronously

Use run_async for non-blocking execution.

import asyncio
from shell_tasks import ShellTaskExecutor

async def demonstrate_async():
    executor = ShellTaskExecutor()
    print("Starting long-running command...")
    task = await executor.run_async("sleep 3 && echo 'Async task finished!'")

    print("Application continues immediately...")
    await asyncio.sleep(1)
    print("Doing other work...")

    result = await task.wait_for_completion()
    print(f"Async result: {result.stdout.strip()}")

# asyncio.run(demonstrate_async())

Setting Environment Variables and Working Directory

Control the execution context precisely.

from shell_tasks import ShellTaskExecutor
import os

executor = ShellTaskExecutor()
# Create a temporary file in a specific directory
os.makedirs("/tmp/my_shell_dir", exist_ok=True)
with open("/tmp/my_shell_dir/test.txt", "w") as f:
    f.write("hello")

result = executor.execute_command(
    "pwd && cat test.txt && echo $MY_CUSTOM_VAR",
    cwd="/tmp/my_shell_dir",
    env={"MY_CUSTOM_VAR": "set_from_app"}
)
print(f"Output:\n{result.stdout.strip()}")
# Expected output (approx):
# /tmp/my_shell_dir
# hello
# set_from_app

Important Considerations

Security

Executing arbitrary shell commands poses significant security risks, especially if commands or their arguments originate from untrusted user input.

• Command Injection: Malicious input can alter the intended command. Always sanitize or validate any user-provided input before incorporating it into a shell command.
• Least Privilege: Run commands with the minimum necessary permissions.
• Avoid shell=True (if applicable): While not directly exposed in the primary API, be aware that underlying subprocess mechanisms often have a shell=True option. Using it can simplify command strings but makes injection vulnerabilities easier to exploit. Shell Tasks abstract this away, providing a safer interface.

Performance

Spawning new processes incurs overhead. For very frequent or performance-critical operations, consider if a native Python implementation or a more direct API integration is more suitable than repeatedly invoking shell commands. Asynchronous execution helps manage responsiveness but does not eliminate the overhead of the external process itself.

Cross-Platform Compatibility

Shell commands are often platform-specific. A command that works on Linux (e.g., ls -l) might behave differently or not exist on Windows (where dir is common). Design commands with cross-platform compatibility in mind or implement platform-specific logic within the application.

Resource Management

Long-running or resource-intensive shell tasks can consume CPU, memory, or file handles. Monitor these tasks and implement timeouts or resource limits where appropriate to prevent resource exhaustion.

Best Practices

• Validate Inputs: Rigorously validate all inputs used to construct shell commands to prevent command injection.
• Use Absolute Paths: Specify absolute paths for executables to avoid reliance on the system's PATH environment variable, which can be inconsistent or manipulated.
• Check Return Codes: Always inspect the return_code of the CommandResult to determine the success or failure of the command. Do not assume success based solely on empty stderr.
• Implement Timeouts: For commands that might hang, use a timeout mechanism (e.g., a timeout parameter if available, or asyncio.wait_for for async tasks) to prevent indefinite blocking.
• Log Outputs: Log stdout and stderr for debugging and auditing purposes, especially for commands executed in production environments.
• Error Handling: Wrap command executions in try-except blocks to catch CommandExecutionError and handle unexpected issues gracefully.