Running Jupyter Notebooks in the Cloud

Jupyter notebooks are the standard for data science workflows. But running them in production—scheduled jobs, API endpoints, automated pipelines—requires executing them without the interactive interface.

HopX sandboxes let you run notebooks programmatically, in isolated environments, with full access to results.

Why Run Notebooks in Sandboxes?

Local notebook execution problems:

Notebooks can execute arbitrary code
User notebooks might contain malicious code
Resource consumption is unpredictable
Environment conflicts between notebooks

Sandbox advantages:

Complete isolation from your infrastructure
Consistent, reproducible environments
Resource limits prevent abuse
Parallel execution at scale

Basic Notebook Execution

Using nbconvert

The simplest way to execute a notebook:

python

from hopx import Sandbox
import json
 
sandbox = Sandbox.create(template="code-interpreter")
 
# Create a sample notebook
notebook = {
    "cells": [
        {
            "cell_type": "code",
            "execution_count": None,
            "metadata": {},
            "outputs": [],
            "source": ["import pandas as pd\n", "import numpy as np\n", "print('Libraries loaded')"]
        },
        {
            "cell_type": "code",
            "execution_count": None,
            "metadata": {},
            "outputs": [],
            "source": ["df = pd.DataFrame({'a': [1,2,3], 'b': [4,5,6]})\n", "print(df)"]
        },
        {
            "cell_type": "code",
            "execution_count": None,
            "metadata": {},
            "outputs": [],
            "source": ["result = df['a'].sum()\n", "print(f'Sum: {result}')"]
        }
    ],
    "metadata": {
        "kernelspec": {
            "display_name": "Python 3",
            "language": "python",
            "name": "python3"
        }
    },
    "nbformat": 4,
    "nbformat_minor": 4
}
 
# Upload notebook
sandbox.files.write("/app/analysis.ipynb", json.dumps(notebook))
 
# Execute with nbconvert
result = sandbox.commands.run("""
cd /app && jupyter nbconvert --to notebook --execute analysis.ipynb --output executed.ipynb
""")
 
print("Execution status:", "Success" if result.exit_code == 0 else "Failed")
print(result.stdout)
 

Extracting Results

After execution, read the output notebook:

python

# Download executed notebook
executed = sandbox.files.read("/app/executed.ipynb")
executed_nb = json.loads(executed)
 
# Extract outputs from each cell
for i, cell in enumerate(executed_nb["cells"]):
    if cell["cell_type"] == "code":
        print(f"\n--- Cell {i} ---")
        print("Source:", "".join(cell["source"][:50]))
        
        for output in cell.get("outputs", []):
            if output["output_type"] == "stream":
                print("Output:", "".join(output["text"]))
            elif output["output_type"] == "execute_result":
                print("Result:", output["data"].get("text/plain", ""))
 

Papermill for Parameterized Notebooks

Papermill is the industry standard for parameterized notebook execution.

Basic Papermill Usage

python

from hopx import Sandbox
import json
 
sandbox = Sandbox.create(template="code-interpreter")
 
# Install papermill
sandbox.commands.run("pip install papermill -q")
 
# Create parameterized notebook
notebook = {
    "cells": [
        {
            "cell_type": "code",
            "metadata": {"tags": ["parameters"]},  # Parameters cell
            "source": ["# Parameters\n", "dataset_path = '/data/default.csv'\n", "threshold = 0.5"]
        },
        {
            "cell_type": "code",
            "metadata": {},
            "source": [
                "import pandas as pd\n",
                "print(f'Loading: {dataset_path}')\n",
                "print(f'Threshold: {threshold}')"
            ]
        }
    ],
    "metadata": {"kernelspec": {"name": "python3"}},
    "nbformat": 4,
    "nbformat_minor": 4
}
 
sandbox.files.write("/app/template.ipynb", json.dumps(notebook))
 
# Execute with parameters
result = sandbox.commands.run("""
papermill /app/template.ipynb /app/output.ipynb \
    -p dataset_path "/data/sales_2024.csv" \
    -p threshold 0.75
""")
 
print(result.stdout)
 

Injecting Complex Parameters

python

# Parameters can be Python objects
params = {
    "config": {
        "model": "random_forest",
        "features": ["age", "income", "score"],
        "hyperparams": {"n_estimators": 100, "max_depth": 10}
    },
    "date_range": ["2024-01-01", "2024-12-31"]
}
 
# Write params as JSON
sandbox.files.write("/app/params.json", json.dumps(params))
 
# Notebook loads params from file
notebook_code = '''
import json
 
with open('/app/params.json') as f:
    params = json.load(f)
 
config = params['config']
date_range = params['date_range']
 
print(f"Model: {config['model']}")
print(f"Features: {config['features']}")
'''
 

Building a Notebook Execution API

FastAPI Notebook Runner

python

from fastapi import FastAPI, UploadFile, HTTPException
from pydantic import BaseModel
from hopx import Sandbox
import json
import base64
 
app = FastAPI()
 
class NotebookRequest(BaseModel):
    notebook: str  # Base64 encoded notebook
    parameters: dict = {}
    timeout: int = 300
 
class NotebookResponse(BaseModel):
    success: bool
    executed_notebook: str  # Base64 encoded
    outputs: list
    error: str = None
 
@app.post("/execute", response_model=NotebookResponse)
async def execute_notebook(request: NotebookRequest):
    sandbox = None
    
    try:
        sandbox = Sandbox.create(template="code-interpreter")
        
        # Decode and upload notebook
        notebook_content = base64.b64decode(request.notebook)
        sandbox.files.write("/app/input.ipynb", notebook_content)
        
        # Write parameters
        sandbox.files.write("/app/params.json", json.dumps(request.parameters))
        
        # Execute
        result = sandbox.commands.run(
            "papermill /app/input.ipynb /app/output.ipynb -f /app/params.json",
            timeout=request.timeout
        )
        
        if result.exit_code != 0:
            return NotebookResponse(
                success=False,
                executed_notebook="",
                outputs=[],
                error=result.stderr
            )
        
        # Read executed notebook
        executed = sandbox.files.read("/app/output.ipynb")
        executed_nb = json.loads(executed)
        
        # Extract outputs
        outputs = []
        for cell in executed_nb["cells"]:
            if cell["cell_type"] == "code":
                cell_outputs = []
                for output in cell.get("outputs", []):
                    cell_outputs.append({
                        "type": output["output_type"],
                        "content": extract_output_content(output)
                    })
                outputs.append(cell_outputs)
        
        return NotebookResponse(
            success=True,
            executed_notebook=base64.b64encode(executed).decode(),
            outputs=outputs
        )
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    
    finally:
        if sandbox:
            sandbox.kill()
 
 
def extract_output_content(output):
    """Extract readable content from notebook output"""
    if output["output_type"] == "stream":
        return "".join(output.get("text", []))
    elif output["output_type"] == "execute_result":
        data = output.get("data", {})
        return data.get("text/plain", data.get("text/html", ""))
    elif output["output_type"] == "display_data":
        data = output.get("data", {})
        if "image/png" in data:
            return {"type": "image", "data": data["image/png"]}
        return data.get("text/plain", "")
    elif output["output_type"] == "error":
        return {
            "ename": output.get("ename"),
            "evalue": output.get("evalue"),
            "traceback": output.get("traceback", [])
        }
    return str(output)
 

Scheduled Notebook Execution

python

import schedule
import time
from hopx import Sandbox
import json
from datetime import datetime
 
class NotebookScheduler:
    def __init__(self):
        self.results = []
    
    def run_notebook(self, notebook_path: str, params: dict):
        """Execute a notebook and store results"""
        sandbox = Sandbox.create(template="code-interpreter")
        
        try:
            # Upload notebook
            with open(notebook_path, "rb") as f:
                sandbox.files.write("/app/notebook.ipynb", f.read())
            
            # Execute
            result = sandbox.commands.run(
                "papermill /app/notebook.ipynb /app/output.ipynb " +
                " ".join([f'-p {k} "{v}"' for k, v in params.items()])
            )
            
            # Store result
            self.results.append({
                "timestamp": datetime.now().isoformat(),
                "notebook": notebook_path,
                "success": result.exit_code == 0,
                "output": result.stdout
            })
            
            # Download output notebook
            output_name = f"output_{datetime.now().strftime('%Y%m%d_%H%M%S')}.ipynb"
            output_data = sandbox.files.read("/app/output.ipynb")
            with open(f"outputs/{output_name}", "wb") as f:
                f.write(output_data)
            
            return result.exit_code == 0
        
        finally:
            sandbox.kill()
    
    def schedule_daily(self, notebook_path: str, params: dict, time_str: str):
        """Schedule notebook to run daily"""
        schedule.every().day.at(time_str).do(
            self.run_notebook, notebook_path, params
        )
    
    def run(self):
        """Start the scheduler"""
        while True:
            schedule.run_pending()
            time.sleep(60)
 
 
# Usage
scheduler = NotebookScheduler()
scheduler.schedule_daily(
    "reports/daily_metrics.ipynb",
    {"date": "today"},
    "06:00"
)
scheduler.run()
 

Handling Notebook Outputs

Extracting Visualizations

python

from hopx import Sandbox
import json
import base64
 
sandbox = Sandbox.create(template="code-interpreter")
 
# Notebook with matplotlib output
notebook = {
    "cells": [
        {
            "cell_type": "code",
            "metadata": {},
            "source": [
                "import matplotlib.pyplot as plt\n",
                "import numpy as np\n",
                "\n",
                "x = np.linspace(0, 10, 100)\n",
                "plt.figure(figsize=(10, 6))\n",
                "plt.plot(x, np.sin(x))\n",
                "plt.title('Sine Wave')\n",
                "plt.show()"
            ]
        }
    ],
    "metadata": {"kernelspec": {"name": "python3"}},
    "nbformat": 4,
    "nbformat_minor": 4
}
 
sandbox.files.write("/app/viz.ipynb", json.dumps(notebook))
 
# Execute
sandbox.commands.run(
    "jupyter nbconvert --to notebook --execute viz.ipynb --output executed.ipynb",
    cwd="/app"
)
 
# Extract images from output
executed = json.loads(sandbox.files.read("/app/executed.ipynb"))
 
images = []
for cell in executed["cells"]:
    for output in cell.get("outputs", []):
        if "image/png" in output.get("data", {}):
            img_data = output["data"]["image/png"]
            images.append(base64.b64decode(img_data))
 
# Save extracted images
for i, img in enumerate(images):
    with open(f"figure_{i}.png", "wb") as f:
        f.write(img)
 
print(f"Extracted {len(images)} images")
 

Capturing DataFrames

python

# Notebook cell that outputs a DataFrame
notebook_code = '''
import pandas as pd
import json
 
df = pd.DataFrame({
    'product': ['A', 'B', 'C'],
    'sales': [100, 150, 80],
    'profit': [20, 45, 15]
})
 
# Output as JSON for programmatic access
print("__DATA_START__")
print(df.to_json(orient='records'))
print("__DATA_END__")
 
# Also display nicely
display(df)
'''
 
# After execution, parse the output
for cell in executed_nb["cells"]:
    for output in cell.get("outputs", []):
        if output["output_type"] == "stream":
            text = "".join(output["text"])
            if "__DATA_START__" in text:
                start = text.index("__DATA_START__") + len("__DATA_START__")
                end = text.index("__DATA_END__")
                json_data = text[start:end].strip()
                data = json.loads(json_data)
                print("Extracted DataFrame:", data)
 

Error Handling

Graceful Notebook Failures

python

from hopx import Sandbox
import json
 
def execute_notebook_safely(notebook_content: bytes, params: dict = None):
    """Execute notebook with comprehensive error handling"""
    sandbox = Sandbox.create(template="code-interpreter")
    
    try:
        sandbox.files.write("/app/notebook.ipynb", notebook_content)
        
        if params:
            sandbox.files.write("/app/params.json", json.dumps(params))
            cmd = "papermill /app/notebook.ipynb /app/output.ipynb -f /app/params.json"
        else:
            cmd = "jupyter nbconvert --execute --to notebook /app/notebook.ipynb --output output.ipynb"
        
        result = sandbox.commands.run(cmd, timeout=300)
        
        # Read output notebook even if execution failed
        try:
            output_nb = json.loads(sandbox.files.read("/app/output.ipynb"))
        except:
            output_nb = None
        
        # Check for cell errors
        errors = []
        if output_nb:
            for i, cell in enumerate(output_nb.get("cells", [])):
                for output in cell.get("outputs", []):
                    if output.get("output_type") == "error":
                        errors.append({
                            "cell": i,
                            "ename": output.get("ename"),
                            "evalue": output.get("evalue"),
                            "traceback": output.get("traceback", [])
                        })
        
        return {
            "success": result.exit_code == 0 and len(errors) == 0,
            "exit_code": result.exit_code,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "notebook": output_nb,
            "errors": errors
        }
    
    except TimeoutError:
        return {
            "success": False,
            "error": "Notebook execution timed out",
            "errors": [{"type": "timeout"}]
        }
    
    except Exception as e:
        return {
            "success": False,
            "error": str(e),
            "errors": [{"type": "exception", "message": str(e)}]
        }
    
    finally:
        sandbox.kill()
 
 
# Usage
with open("analysis.ipynb", "rb") as f:
    result = execute_notebook_safely(f.read(), {"threshold": 0.5})
 
if not result["success"]:
    print("Execution failed!")
    for error in result["errors"]:
        print(f"  Cell {error.get('cell', '?')}: {error.get('evalue', error)}")
 

Parallel Notebook Execution

Run multiple notebooks simultaneously:

python

from hopx import Sandbox
import concurrent.futures
import json
 
def run_single_notebook(notebook_path: str, params: dict):
    """Run one notebook in its own sandbox"""
    sandbox = Sandbox.create(template="code-interpreter")
    
    try:
        with open(notebook_path, "rb") as f:
            sandbox.files.write("/app/notebook.ipynb", f.read())
        
        sandbox.files.write("/app/params.json", json.dumps(params))
        
        result = sandbox.commands.run(
            "papermill /app/notebook.ipynb /app/output.ipynb -f /app/params.json",
            timeout=300
        )
        
        output = sandbox.files.read("/app/output.ipynb")
        
        return {
            "notebook": notebook_path,
            "params": params,
            "success": result.exit_code == 0,
            "output": output
        }
    
    finally:
        sandbox.kill()
 
 
def run_notebooks_parallel(jobs: list[tuple[str, dict]], max_workers: int = 5):
    """Run multiple notebooks in parallel"""
    results = []
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(run_single_notebook, path, params): (path, params)
            for path, params in jobs
        }
        
        for future in concurrent.futures.as_completed(futures):
            path, params = futures[future]
            try:
                result = future.result()
                results.append(result)
                print(f"✓ Completed: {path}")
            except Exception as e:
                results.append({
                    "notebook": path,
                    "params": params,
                    "success": False,
                    "error": str(e)
                })
                print(f"✗ Failed: {path} - {e}")
    
    return results
 
 
# Run 10 notebooks with different parameters
jobs = [
    ("analysis.ipynb", {"region": region})
    for region in ["US", "EU", "APAC", "LATAM", "MEA"]
]
 
results = run_notebooks_parallel(jobs, max_workers=5)
print(f"Completed: {sum(1 for r in results if r['success'])}/{len(results)}")
 

Best Practices

1. Version Your Notebooks

python

# Add version metadata
notebook["metadata"]["version"] = "1.2.0"
notebook["metadata"]["last_modified"] = "2024-11-15"
 
# Check version before execution
if notebook["metadata"].get("version", "0.0.0") < "1.0.0":
    raise ValueError("Notebook version too old")
 

2. Validate Inputs

python

def validate_notebook(content: bytes) -> bool:
    """Validate notebook before execution"""
    try:
        nb = json.loads(content)
        
        # Check format
        if nb.get("nbformat", 0) < 4:
            raise ValueError("Notebook format too old")
        
        # Check for required cells
        has_code = any(c["cell_type"] == "code" for c in nb["cells"])
        if not has_code:
            raise ValueError("Notebook has no code cells")
        
        # Check for dangerous imports
        dangerous = ["os.system", "subprocess", "eval(", "exec("]
        for cell in nb["cells"]:
            source = "".join(cell.get("source", []))
            for d in dangerous:
                if d in source:
                    raise ValueError(f"Potentially dangerous code: {d}")
        
        return True
    
    except json.JSONDecodeError:
        raise ValueError("Invalid JSON")
 

3. Resource Limits

python

# Set memory and time limits
sandbox.commands.run(
    "timeout 300 jupyter nbconvert --execute notebook.ipynb",
    memory_limit="4G"
)
 

4. Clean Output Notebooks

python

def clean_notebook_output(notebook: dict) -> dict:
    """Remove outputs for storage/version control"""
    cleaned = notebook.copy()
    
    for cell in cleaned["cells"]:
        if cell["cell_type"] == "code":
            cell["outputs"] = []
            cell["execution_count"] = None
    
    return cleaned
 

Conclusion

Running Jupyter notebooks in HopX sandboxes enables:

Safe execution of untrusted notebooks
Scalable automation with parallel execution
Consistent environments for reproducibility
Programmatic access to notebook outputs

Whether you're building notebook-powered APIs, scheduling data science jobs, or processing user-submitted notebooks, sandboxed execution is the secure, scalable solution.

Running Jupyter Notebooks in the Cloud

Running Jupyter Notebooks in the Cloud

Why Run Notebooks in Sandboxes?

Basic Notebook Execution

Using nbconvert

Extracting Results

Papermill for Parameterized Notebooks

Basic Papermill Usage

Injecting Complex Parameters

Building a Notebook Execution API

FastAPI Notebook Runner

Scheduled Notebook Execution

Handling Notebook Outputs

Extracting Visualizations

Capturing DataFrames

Error Handling

Graceful Notebook Failures

Parallel Notebook Execution

Best Practices

1. Version Your Notebooks

2. Validate Inputs

3. Resource Limits

4. Clean Output Notebooks

Conclusion

Resources

Related articles

Microsoft Agent Framework with HopX: Secure Code Execution for AI Agents

Microsoft AutoGen with Isolated Code Execution Using HopX

CrewAI Multi-Agent Pipelines with Secure Code Execution

1	from hopx import Sandbox
2	import json
3
4	sandbox = Sandbox.create(template="code-interpreter")
5
6	# Create a sample notebook
7	notebook = {
8	"cells": [
9	{
10	"cell_type": "code",
11	"execution_count": None,
12	"metadata": {},
13	"outputs": [],
14	"source": ["import pandas as pd\n", "import numpy as np\n", "print('Libraries loaded')"]
15	},
16	{
17	"cell_type": "code",
18	"execution_count": None,
19	"metadata": {},
20	"outputs": [],
21	"source": ["df = pd.DataFrame({'a': [1,2,3], 'b': [4,5,6]})\n", "print(df)"]
22	},
23	{
24	"cell_type": "code",
25	"execution_count": None,
26	"metadata": {},
27	"outputs": [],
28	"source": ["result = df['a'].sum()\n", "print(f'Sum: {result}')"]
29	}
30	],
31	"metadata": {
32	"kernelspec": {
33	"display_name": "Python 3",
34	"language": "python",
35	"name": "python3"
36	}
37	},
38	"nbformat": 4,
39	"nbformat_minor": 4
40	}
41
42	# Upload notebook
43	sandbox.files.write("/app/analysis.ipynb", json.dumps(notebook))
44
45	# Execute with nbconvert
46	result = sandbox.commands.run("""
47	cd /app && jupyter nbconvert --to notebook --execute analysis.ipynb --output executed.ipynb
48	""")
49
50	print("Execution status:", "Success" if result.exit_code == 0 else "Failed")
51	print(result.stdout)
52

1	# Download executed notebook
2	executed = sandbox.files.read("/app/executed.ipynb")
3	executed_nb = json.loads(executed)
4
5	# Extract outputs from each cell
6	for i, cell in enumerate(executed_nb["cells"]):
7	if cell["cell_type"] == "code":
8	print(f"\n--- Cell {i} ---")
9	print("Source:", "".join(cell["source"][:50]))
10
11	for output in cell.get("outputs", []):
12	if output["output_type"] == "stream":
13	print("Output:", "".join(output["text"]))
14	elif output["output_type"] == "execute_result":
15	print("Result:", output["data"].get("text/plain", ""))
16

1	# Parameters can be Python objects
2	params = {
3	"config": {
4	"model": "random_forest",
5	"features": ["age", "income", "score"],
6	"hyperparams": {"n_estimators": 100, "max_depth": 10}
7	},
8	"date_range": ["2024-01-01", "2024-12-31"]
9	}
10
11	# Write params as JSON
12	sandbox.files.write("/app/params.json", json.dumps(params))
13
14	# Notebook loads params from file
15	notebook_code = '''
16	import json
17
18	with open('/app/params.json') as f:
19	params = json.load(f)
20
21	config = params['config']
22	date_range = params['date_range']
23
24	print(f"Model: {config['model']}")
25	print(f"Features: {config['features']}")
26	'''
27

1	from fastapi import FastAPI, UploadFile, HTTPException
2	from pydantic import BaseModel
3	from hopx import Sandbox
4	import json
5	import base64
6
7	app = FastAPI()
8
9	class NotebookRequest(BaseModel):
10	notebook: str # Base64 encoded notebook
11	parameters: dict = {}
12	timeout: int = 300
13
14	class NotebookResponse(BaseModel):
15	success: bool
16	executed_notebook: str # Base64 encoded
17	outputs: list
18	error: str = None
19
20	@app.post("/execute", response_model=NotebookResponse)
21	async def execute_notebook(request: NotebookRequest):
22	sandbox = None
23
24	try:
25	sandbox = Sandbox.create(template="code-interpreter")
26
27	# Decode and upload notebook
28	notebook_content = base64.b64decode(request.notebook)
29	sandbox.files.write("/app/input.ipynb", notebook_content)
30
31	# Write parameters
32	sandbox.files.write("/app/params.json", json.dumps(request.parameters))
33
34	# Execute
35	result = sandbox.commands.run(
36	"papermill /app/input.ipynb /app/output.ipynb -f /app/params.json",
37	timeout=request.timeout
38	)
39
40	if result.exit_code != 0:
41	return NotebookResponse(
42	success=False,
43	executed_notebook="",
44	outputs=[],
45	error=result.stderr
46	)
47
48	# Read executed notebook
49	executed = sandbox.files.read("/app/output.ipynb")
50	executed_nb = json.loads(executed)
51
52	# Extract outputs
53	outputs = []
54	for cell in executed_nb["cells"]:
55	if cell["cell_type"] == "code":
56	cell_outputs = []
57	for output in cell.get("outputs", []):
58	cell_outputs.append({
59	"type": output["output_type"],
60	"content": extract_output_content(output)
61	})
62	outputs.append(cell_outputs)
63
64	return NotebookResponse(
65	success=True,
66	executed_notebook=base64.b64encode(executed).decode(),
67	outputs=outputs
68	)
69
70	except Exception as e:
71	raise HTTPException(status_code=500, detail=str(e))
72
73	finally:
74	if sandbox:
75	sandbox.kill()
76
77
78	def extract_output_content(output):
79	"""Extract readable content from notebook output"""
80	if output["output_type"] == "stream":
81	return "".join(output.get("text", []))
82	elif output["output_type"] == "execute_result":
83	data = output.get("data", {})
84	return data.get("text/plain", data.get("text/html", ""))
85	elif output["output_type"] == "display_data":
86	data = output.get("data", {})
87	if "image/png" in data:
88	return {"type": "image", "data": data["image/png"]}
89	return data.get("text/plain", "")
90	elif output["output_type"] == "error":
91	return {
92	"ename": output.get("ename"),
93	"evalue": output.get("evalue"),
94	"traceback": output.get("traceback", [])
95	}
96	return str(output)
97

1	import schedule
2	import time
3	from hopx import Sandbox
4	import json
5	from datetime import datetime
6
7	class NotebookScheduler:
8	def __init__(self):
9	self.results = []
10
11	def run_notebook(self, notebook_path: str, params: dict):
12	"""Execute a notebook and store results"""
13	sandbox = Sandbox.create(template="code-interpreter")
14
15	try:
16	# Upload notebook
17	with open(notebook_path, "rb") as f:
18	sandbox.files.write("/app/notebook.ipynb", f.read())
19
20	# Execute
21	result = sandbox.commands.run(
22	"papermill /app/notebook.ipynb /app/output.ipynb " +
23	" ".join([f'-p {k} "{v}"' for k, v in params.items()])
24	)
25
26	# Store result
27	self.results.append({
28	"timestamp": datetime.now().isoformat(),
29	"notebook": notebook_path,
30	"success": result.exit_code == 0,
31	"output": result.stdout
32	})
33
34	# Download output notebook
35	output_name = f"output_{datetime.now().strftime('%Y%m%d_%H%M%S')}.ipynb"
36	output_data = sandbox.files.read("/app/output.ipynb")
37	with open(f"outputs/{output_name}", "wb") as f:
38	f.write(output_data)
39
40	return result.exit_code == 0
41
42	finally:
43	sandbox.kill()
44
45	def schedule_daily(self, notebook_path: str, params: dict, time_str: str):
46	"""Schedule notebook to run daily"""
47	schedule.every().day.at(time_str).do(
48	self.run_notebook, notebook_path, params
49	)
50
51	def run(self):
52	"""Start the scheduler"""
53	while True:
54	schedule.run_pending()
55	time.sleep(60)
56
57
58	# Usage
59	scheduler = NotebookScheduler()
60	scheduler.schedule_daily(
61	"reports/daily_metrics.ipynb",
62	{"date": "today"},
63	"06:00"
64	)
65	scheduler.run()
66

1	# Notebook cell that outputs a DataFrame
2	notebook_code = '''
3	import pandas as pd
4	import json
5
6	df = pd.DataFrame({
7	'product': ['A', 'B', 'C'],
8	'sales': [100, 150, 80],
9	'profit': [20, 45, 15]
10	})
11
12	# Output as JSON for programmatic access
13	print("__DATA_START__")
14	print(df.to_json(orient='records'))
15	print("__DATA_END__")
16
17	# Also display nicely
18	display(df)
19	'''
20
21	# After execution, parse the output
22	for cell in executed_nb["cells"]:
23	for output in cell.get("outputs", []):
24	if output["output_type"] == "stream":
25	text = "".join(output["text"])
26	if "__DATA_START__" in text:
27	start = text.index("__DATA_START__") + len("__DATA_START__")
28	end = text.index("__DATA_END__")
29	json_data = text[start:end].strip()
30	data = json.loads(json_data)
31	print("Extracted DataFrame:", data)
32

1	from hopx import Sandbox
2	import json
3
4	def execute_notebook_safely(notebook_content: bytes, params: dict = None):
5	"""Execute notebook with comprehensive error handling"""
6	sandbox = Sandbox.create(template="code-interpreter")
7
8	try:
9	sandbox.files.write("/app/notebook.ipynb", notebook_content)
10
11	if params:
12	sandbox.files.write("/app/params.json", json.dumps(params))
13	cmd = "papermill /app/notebook.ipynb /app/output.ipynb -f /app/params.json"
14	else:
15	cmd = "jupyter nbconvert --execute --to notebook /app/notebook.ipynb --output output.ipynb"
16
17	result = sandbox.commands.run(cmd, timeout=300)
18
19	# Read output notebook even if execution failed
20	try:
21	output_nb = json.loads(sandbox.files.read("/app/output.ipynb"))
22	except:
23	output_nb = None
24
25	# Check for cell errors
26	errors = []
27	if output_nb:
28	for i, cell in enumerate(output_nb.get("cells", [])):
29	for output in cell.get("outputs", []):
30	if output.get("output_type") == "error":
31	errors.append({
32	"cell": i,
33	"ename": output.get("ename"),
34	"evalue": output.get("evalue"),
35	"traceback": output.get("traceback", [])
36	})
37
38	return {
39	"success": result.exit_code == 0 and len(errors) == 0,
40	"exit_code": result.exit_code,
41	"stdout": result.stdout,
42	"stderr": result.stderr,
43	"notebook": output_nb,
44	"errors": errors
45	}
46
47	except TimeoutError:
48	return {
49	"success": False,
50	"error": "Notebook execution timed out",
51	"errors": [{"type": "timeout"}]
52	}
53
54	except Exception as e:
55	return {
56	"success": False,
57	"error": str(e),
58	"errors": [{"type": "exception", "message": str(e)}]
59	}
60
61	finally:
62	sandbox.kill()
63
64
65	# Usage
66	with open("analysis.ipynb", "rb") as f:
67	result = execute_notebook_safely(f.read(), {"threshold": 0.5})
68
69	if not result["success"]:
70	print("Execution failed!")
71	for error in result["errors"]:
72	print(f" Cell {error.get('cell', '?')}: {error.get('evalue', error)}")
73

1	from hopx import Sandbox
2	import concurrent.futures
3	import json
4
5	def run_single_notebook(notebook_path: str, params: dict):
6	"""Run one notebook in its own sandbox"""
7	sandbox = Sandbox.create(template="code-interpreter")
8
9	try:
10	with open(notebook_path, "rb") as f:
11	sandbox.files.write("/app/notebook.ipynb", f.read())
12
13	sandbox.files.write("/app/params.json", json.dumps(params))
14
15	result = sandbox.commands.run(
16	"papermill /app/notebook.ipynb /app/output.ipynb -f /app/params.json",
17	timeout=300
18	)
19
20	output = sandbox.files.read("/app/output.ipynb")
21
22	return {
23	"notebook": notebook_path,
24	"params": params,
25	"success": result.exit_code == 0,
26	"output": output
27	}
28
29	finally:
30	sandbox.kill()
31
32
33	def run_notebooks_parallel(jobs: list[tuple[str, dict]], max_workers: int = 5):
34	"""Run multiple notebooks in parallel"""
35	results = []
36
37	with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
38	futures = {
39	executor.submit(run_single_notebook, path, params): (path, params)
40	for path, params in jobs
41	}
42
43	for future in concurrent.futures.as_completed(futures):
44	path, params = futures[future]
45	try:
46	result = future.result()
47	results.append(result)
48	print(f"✓ Completed: {path}")
49	except Exception as e:
50	results.append({
51	"notebook": path,
52	"params": params,
53	"success": False,
54	"error": str(e)
55	})
56	print(f"✗ Failed: {path} - {e}")
57
58	return results
59
60
61	# Run 10 notebooks with different parameters
62	jobs = [
63	("analysis.ipynb", {"region": region})
64	for region in ["US", "EU", "APAC", "LATAM", "MEA"]
65	]
66
67	results = run_notebooks_parallel(jobs, max_workers=5)
68	print(f"Completed: {sum(1 for r in results if r['success'])}/{len(results)}")
69

1	# Add version metadata
2	notebook["metadata"]["version"] = "1.2.0"
3	notebook["metadata"]["last_modified"] = "2024-11-15"
4
5	# Check version before execution
6	if notebook["metadata"].get("version", "0.0.0") < "1.0.0":
7	raise ValueError("Notebook version too old")
8

1	def validate_notebook(content: bytes) -> bool:
2	"""Validate notebook before execution"""
3	try:
4	nb = json.loads(content)
5
6	# Check format
7	if nb.get("nbformat", 0) < 4:
8	raise ValueError("Notebook format too old")
9
10	# Check for required cells
11	has_code = any(c["cell_type"] == "code" for c in nb["cells"])
12	if not has_code:
13	raise ValueError("Notebook has no code cells")
14
15	# Check for dangerous imports
16	dangerous = ["os.system", "subprocess", "eval(", "exec("]
17	for cell in nb["cells"]:
18	source = "".join(cell.get("source", []))
19	for d in dangerous:
20	if d in source:
21	raise ValueError(f"Potentially dangerous code: {d}")
22
23	return True
24
25	except json.JSONDecodeError:
26	raise ValueError("Invalid JSON")
27

1	# Set memory and time limits
2	sandbox.commands.run(
3	"timeout 300 jupyter nbconvert --execute notebook.ipynb",
4	memory_limit="4G"
5	)
6

1	def clean_notebook_output(notebook: dict) -> dict:
2	"""Remove outputs for storage/version control"""
3	cleaned = notebook.copy()
4
5	for cell in cleaned["cells"]:
6	if cell["cell_type"] == "code":
7	cell["outputs"] = []
8	cell["execution_count"] = None
9
10	return cleaned
11