ExecutionResult & Output Types

import runtimed

with runtimed.Session() as session:
    session.start_kernel()
    
    # Execute code
    result = session.run("print('hello')")
    
    # Check status
    print(result.success)         # True
    print(result.execution_count) # 1
    
    # Access outputs
    print(result.stdout)          # "hello\n"
    print(len(result.outputs))    # 1

result = session.run("raise ValueError('oops')")

print(result.success)  # False

if result.error:
    print(f"Error: {result.error.ename}")   # "ValueError"
    print(f"Message: {result.error.evalue}") # "oops"
    for line in result.error.traceback:
        print(line)

for output in result.outputs:
    if output.output_type == "stream":
        print(f"{output.name}: {output.text}")

result = session.run("42")

for output in result.display_data:
    if "text/plain" in output.data:
        print(output.data["text/plain"])  # "42"

result = session.run("1 / 0")

if result.error:
    print(f"Exception: {result.error.ename}")    # "ZeroDivisionError"
    print(f"Message: {result.error.evalue}")     # "division by zero"
    print(f"Traceback has {len(result.error.traceback)} lines")

cell = session.get_cell(cell_id)

print(f"ID: {cell.id}")
print(f"Type: {cell.cell_type}")
print(f"Source: {cell.source}")
print(f"Executed: {cell.execution_count is not None}")

import runtimed

with runtimed.Session() as session:
    session.start_kernel()
    
    # Execute code with multiple outputs
    code = """
import matplotlib.pyplot as plt
print("Generating plot...")
plt.plot([1, 2, 3, 4])
plt.show()
print("Done")
"""
    
    result = session.run(code)
    
    # Process each output type
    for output in result.outputs:
        if output.output_type == "stream":
            print(f"[{output.name}] {output.text}")
        
        elif output.output_type == "display_data":
            print(f"Display output with MIME types: {list(output.data.keys())}")
            if "image/png" in output.data:
                # Save the image
                import base64
                png_data = base64.b64decode(output.data["image/png"])
                with open("plot.png", "wb") as f:
                    f.write(png_data)
        
        elif output.output_type == "error":
            print(f"Error: {output.ename}: {output.evalue}")

import runtimed

with runtimed.Session() as session:
    session.start_kernel()
    
    cells_to_run = [
        "import pandas as pd",
        "df = pd.DataFrame({'a': [1, 2, 3]})",
        "print(df.sum())"
    ]
    
    for code in cells_to_run:
        result = session.run(code)
        
        if not result.success:
            print(f"Execution failed: {code}")
            if result.error:
                print(f"  Error: {result.error.ename}: {result.error.evalue}")
            break
        
        if result.stdout:
            print(f"Output: {result.stdout}")

import asyncio
import runtimed

async def run_and_process(session, code):
    result = await session.run(code)
    
    return {
        "code": code,
        "success": result.success,
        "stdout": result.stdout,
        "stderr": result.stderr,
        "execution_count": result.execution_count
    }

async def main():
    async with runtimed.AsyncSession() as session:
        await session.start_kernel()
        
        # Run multiple cells in parallel
        tasks = [
            run_and_process(session, "print('Task 1')"),
            run_and_process(session, "print('Task 2')"),
            run_and_process(session, "print('Task 3')")
        ]
        
        results = await asyncio.gather(*tasks)
        
        for result in results:
            print(f"Code: {result['code']}")
            print(f"  Success: {result['success']}")
            print(f"  Output: {result['stdout']}")

asyncio.run(main())

result = session.run("""
import pandas as pd
df = pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]})
df
""")

# Get the table as HTML
for output in result.display_data:
    if "text/html" in output.data:
        html = output.data["text/html"]
        with open("table.html", "w") as f:
            f.write(html)
    
    if "text/plain" in output.data:
        print(output.data["text/plain"])

import base64
from pathlib import Path

def save_display_outputs(result, output_dir="outputs"):
    """Save all display outputs to files."""
    output_dir = Path(output_dir)
    output_dir.mkdir(exist_ok=True)
    
    for i, output in enumerate(result.display_data):
        for mime_type, content in output.data.items():
            if mime_type == "image/png":
                png_data = base64.b64decode(content)
                (output_dir / f"output_{i}.png").write_bytes(png_data)
            
            elif mime_type == "text/html":
                (output_dir / f"output_{i}.html").write_text(content)
            
            elif mime_type == "application/json":
                (output_dir / f"output_{i}.json").write_text(content)

# Usage
result = session.run("plt.plot([1,2,3]); plt.show()")
save_display_outputs(result)

result = session.run("""
def divide(a, b):
    return a / b

divide(10, 0)
""")

if result.error:
    print("Error occurred:")
    print(f"  Type: {result.error.ename}")
    print(f"  Message: {result.error.evalue}")
    print("\nTraceback:")
    for line in result.error.traceback:
        # Traceback lines include ANSI color codes
        print(line)

def execute_with_retry(session, code, max_retries=3):
    """Execute code with retry on specific errors."""
    for attempt in range(max_retries):
        result = session.run(code)
        
        if result.success:
            return result
        
        if result.error:
            # Don't retry on syntax errors
            if result.error.ename == "SyntaxError":
                raise RuntimeError(f"Syntax error: {result.error.evalue}")
            
            # Retry on other errors
            print(f"Attempt {attempt + 1} failed: {result.error.ename}")
            if attempt == max_retries - 1:
                raise RuntimeError(f"Max retries exceeded: {result.error.evalue}")
    
    return result

# Usage
try:
    result = execute_with_retry(session, "import flaky_module")
except RuntimeError as e:
    print(f"Failed: {e}")