The Sandbox provides process management through the process property. You can execute Python code, run shell commands, and manage long-running processes with real-time output streaming.

​
Running Python Code

​
Basic Code Execution

from beam import Sandbox, Image

sandbox = Sandbox(image=Image(python_version=PythonVersion.Python311))
sb = sandbox.create()

# Run simple Python code
result = sb.process.run_code("print('Hello from sandbox!')")
print(result.result)  # Hello from sandbox!
print(f"Exit code: {result.exit_code}")  # 0

​
Complex Python Scripts

# Multi-line Python code
code = """
import numpy as np
import pandas as pd

# Generate sample data
data = np.random.randn(1000, 3)
df = pd.DataFrame(data, columns=['A', 'B', 'C'])

# Calculate statistics
stats = df.describe()
print("Data Statistics:")
print(stats)

# Save results
df.to_csv('/workspace/data.csv', index=False)
print("Data saved to /workspace/data.csv")
"""

response = sb.process.run_code(code)
print(response.result)

​
Error Handling

# Code with errors
response = sb.process.run_code("""
import nonexistent_module
print("This won't execute")
""")

print(f"Exit code: {response.exit_code}")  # Non-zero exit code
print(f"Error output: {response.result}")  # Error message

​
Executing Commands

​
Basic Command Execution

# Run a simple command
process = sb.process.exec("ls", "-la", "/workspace")

# Wait for completion
exit_code = process.wait()
print(f"Command completed with exit code: {exit_code}")

# Get all output
for line in process.logs:
    print(line, end="")

​
Shell Commands

# Use shell features
process = sb.process.exec("echo $HOME && pwd && whoami")

# Wait and get output
process.wait()
for line in process.logs:
    print(line, end="")

​
Working Directory

# Execute in specific directory
process = sb.process.exec("ls", "-la", cwd="/workspace")
process.wait()

# Create directory and work in it
sb.process.run_code("import os; os.makedirs('/workspace/myproject', exist_ok=True)")
process = sb.process.exec("touch", "test.txt", cwd="/workspace/myproject")

​
Environment Variables

# Set environment variables for command
env = {
    "DATABASE_URL": "postgresql://localhost/mydb",
    "DEBUG": "true",
    "API_KEY": "secret-key"
}

process = sb.process.exec("env", "|", "grep", "DATABASE", env=env)
process.wait()

​
Non-blocking Execution

​
Background Processes

# Start a long-running process without waiting
process = sb.process.run_code("""
import time
for i in range(10):
    print(f"Processing {i}...")
    time.sleep(1)
""", blocking=False)

print(f"Process started with PID: {process.pid}")

# Do other work while it runs
print("Process is running in background...")

# Check if still running
print(f"Exit code: {process.exit_code}")  # -1 if still running

# Wait for completion when ready
process.wait()
print("Process completed!")

​
Real-time Output Streaming

# Start process and stream output in real-time
process = sb.process.run_code("""
import time
for i in range(5):
    print(f"Step {i}: Processing...")
    time.sleep(1)
print("Done!")
""", blocking=False)

# Stream output as it happens
for line in process.logs:
    print(f"[REAL-TIME] {line}", end="")

​
Process Control

​
Process Management

# Start multiple processes
process1 = sb.process.exec("sleep", "30", blocking=False)
process2 = sb.process.exec("sleep", "60", blocking=False)

print(f"Process 1 PID: {process1.pid}")
print(f"Process 2 PID: {process2.pid}")

# List all running processes
for p in sb.process.list_processes():
    print(f"PID {p.pid}: {p.status()}")

# Kill specific process
process1.kill()
print("Process 1 killed")

# Get process by PID
specific_process = sb.process.get_process(process2.pid)
print(f"Process 2 status: {specific_process.status()}")

​
Process Status and Monitoring

# Start a process
process = sb.process.exec("sleep", "10", blocking=False)

# Monitor status
while True:
    exit_code, status = process.status()
    print(f"PID {process.pid}: Exit code {exit_code}, Status: {status}")
    
    if exit_code >= 0:
        print("Process completed")
        break
    
    time.sleep(1)

​
Process Output Streams

# Start process with output
process = sb.process.run_code("""
import sys
print("This goes to stdout")
print("This also goes to stdout", file=sys.stdout)
print("This goes to stderr", file=sys.stderr)
""", blocking=False)

# Read stdout
print("=== STDOUT ===")
for line in process.stdout:
    print(f"STDOUT: {line}", end="")

# Read stderr
print("=== STDERR ===")
for line in process.stderr:
    print(f"STDERR: {line}", end="")

# Read combined logs
print("=== COMBINED LOGS ===")
for line in process.logs:
    print(f"LOG: {line}", end="")