command.py•5.8 kB
#!/usr/bin/env python3
"""
Example demonstrating how to use sandbox.command.run to execute shell commands.
"""
import asyncio
import aiohttp
from microsandbox import PythonSandbox
async def basic_example():
"""Example showing basic command execution with context manager."""
print("\n=== Basic Command Example ===")
# Create a sandbox using a context manager (automatically handles start/stop)
async with PythonSandbox.create(name="command-example") as sandbox:
# Run a simple command
ls_execution = await sandbox.command.run("ls", ["-la", "/"])
print("$ ls -la /")
print(f"Exit code: {ls_execution.exit_code}")
print("Output:")
print(await ls_execution.output())
# Execute a command with string arguments
echo_execution = await sandbox.command.run(
"echo", ["Hello from", "sandbox command!"]
)
print("\n$ echo Hello from sandbox command!")
print(f"Output: {await echo_execution.output()}")
# Get system information
uname_execution = await sandbox.command.run("uname", ["-a"])
print("\n$ uname -a")
print(f"Output: {await uname_execution.output()}")
async def error_handling_example():
"""Example showing how to handle command errors."""
print("\n=== Error Handling Example ===")
async with PythonSandbox.create(name="error-example") as sandbox:
# Run a command that generates an error
error_execution = await sandbox.command.run("ls", ["/nonexistent"])
print("$ ls /nonexistent")
print(f"Exit code: {error_execution.exit_code}")
print(f"Success: {error_execution.success}")
print("Error output:")
print(await error_execution.error())
# Deliberately cause a command not found error
try:
_nonexistent_cmd = await sandbox.command.run("nonexistentcommand", [])
# This should not execute if the command fails
print("Command succeeded unexpectedly")
except RuntimeError as e:
print(f"\nCaught exception for nonexistent command: {e}")
async def timeout_example():
"""Example showing how to use command timeouts."""
print("\n=== Timeout Example ===")
async with PythonSandbox.create(name="timeout-example") as sandbox:
print("Running command with timeout...")
try:
# Run a command that takes longer than the specified timeout
await sandbox.command.run("sleep", ["10"], timeout=2)
print("Command completed (unexpected!)")
except RuntimeError as e:
print(f"Command timed out as expected: {e}")
# Show that the sandbox is still usable after a timeout
echo_execution = await sandbox.command.run("echo", ["Still working!"])
print(f"\nSandbox still works: {await echo_execution.output()}")
async def advanced_example():
"""Example showing more advanced command usage."""
print("\n=== Advanced Example ===")
async with PythonSandbox.create(name="advanced-example") as sandbox:
# Write a file
write_cmd = await sandbox.command.run(
"bash", ["-c", "echo 'Hello, file content!' > /tmp/test.txt"]
)
print(f"Created file, exit code: {write_cmd.exit_code}")
# Read the file back
read_cmd = await sandbox.command.run("cat", ["/tmp/test.txt"])
print(f"File content: {await read_cmd.output()}")
# Run a more complex pipeline
pipeline_cmd = await sandbox.command.run(
"bash",
[
"-c",
"mkdir -p /tmp/test_dir && "
"echo 'Line 1' > /tmp/test_dir/data.txt && "
"echo 'Line 2' >> /tmp/test_dir/data.txt && "
"cat /tmp/test_dir/data.txt | grep 'Line' | wc -l",
],
)
print(f"\nPipeline output (should be 2): {await pipeline_cmd.output()}")
# Create and run a Python script
create_script = await sandbox.command.run(
"bash",
[
"-c",
"""cat > /tmp/test.py << 'EOF'
import sys
print("Python script executed!")
print(f"Arguments: {sys.argv[1:]}")
EOF""",
],
)
if create_script.success:
# Run the script with arguments
script_cmd = await sandbox.command.run(
"python", ["/tmp/test.py", "arg1", "arg2", "arg3"]
)
print("\nPython script output:")
print(await script_cmd.output())
async def explicit_lifecycle_example():
"""Example showing explicit lifecycle management."""
print("\n=== Explicit Lifecycle Example ===")
# Create sandbox without context manager
sandbox = PythonSandbox(name="explicit-lifecycle")
sandbox._session = aiohttp.ClientSession()
try:
# Manually start the sandbox
print("Starting sandbox...")
await sandbox.start()
# Execute commands
hostname_cmd = await sandbox.command.run("hostname")
print(f"Hostname: {await hostname_cmd.output()}")
date_cmd = await sandbox.command.run("date")
print(f"Date: {await date_cmd.output()}")
finally:
# Manually stop the sandbox and close session
print("Stopping sandbox...")
await sandbox.stop()
await sandbox._session.close()
async def main():
"""Main function to run all examples."""
print("Command Execution Examples")
print("=========================")
await basic_example()
await error_handling_example()
await timeout_example()
await advanced_example()
await explicit_lifecycle_example()
print("\nAll examples completed!")
if __name__ == "__main__":
asyncio.run(main())