from contextlib import redirect_stdout
from io import StringIO
from fastcore.nbio import read_nb
from fastcore.test import test_eq
from nbskill.execute import exec_nb as _example_exec_nb
from nbskill.write import write_nb as _example_write_nb
from nbskill.foundation import demo_path, remove_demo_path, write_demo_notebook, write_tool_notebook
from nbskill.write import update_cell, write_nbExecuting notebooks
Execute notebooks with visible outputs, local package imports, and a fast per-cell timeout. Cells that exceed the timeout are marked with execution-fingerprint metadata and skipped on later runs until their source changes.
Execution closes the loop after reading and writing. The project needs a way to run notebooks as notebooks, with local imports available and with visible outputs copied back into the notebook for inspection.
This notebook wraps execnb with nbdev-friendly defaults: local import paths, optional partial execution, timeout handling, and a test helper that reports notebook errors in a concise form.
Execution is the confidence step after an edit. The wrapper keeps imports close to how nbdev users run notebooks, records visible outputs for inspection, and marks timed-out cells so repeated runs do not keep blocking on the same long operation.
exec_nb("nbs/03_execute.ipynb", up2id="some-cell-id", timeout=10, show_output=True)Production contract
Notebook execution is production core. exec_nb must run notebooks with local project imports available, support partial execution by cell or chapter, capture visible output, keep safe mode on by default, mark timed-out cells so repeated runs do not trap users, and recover when timed-out source changes.
with write_demo_notebook("03_execute_example.ipynb") as path:
_example_write_nb(str(path), "%%code\nvalue = 6 * 7\nprint(value)", replace=True)
_example_exec_nb(str(path), timeout=5, show_output=True)Fastcore 2 removes fastcore.script.Param, and current safepyrun still imports it at module import time. Keep safepyrun lazy so basic discovery and non-execution tools remain available without this optional compatibility failure.
Choosing how much to run
Sometimes verification only needs the first few cells or a single chapter. These helpers translate an index, a cell id, or a chapter into pre/post hooks that stop execution at the right point.
Importing like the project does
Notebook execution should see the same local package that tests and examples see. This section finds the project root from common markers and puts the notebook folder, root, and optional src folder on the shell path.
Safe execution also keeps a project policy file named .nbskill-exec-approval.json. Each run syncs top-level function names from the notebook into that file. Functions with no statically detected external effects default to allowed: true; functions that look like they can write files, spawn processes, remove files, or make network calls default to allowed: false until the project file is edited.
Keeping project approvals current
_sync_execution_policy removes an approval only when the entry’s recorded cell no longer exists in its notebook. This keeps deleted cells from accumulating without removing approvals owned by other notebooks.
Safepyrun executes statements with copies of the notebook globals, so notebook functions and methods need later bindings synchronized. Only Python functions carrying this shell’s unique marker may be updated. Updating imported callables corrupts their module globals, including dataclasses’ resolution of KW_ONLY; callable classes such as FunctionType expose a __globals__ descriptor rather than a dictionary.
Safepyrun 0.2 defaults to blocking def statements before nbskill’s project policy can approve them. We keep definitions enabled in the runner and let the notebook execution policy decide which functions are allowed to run.
NBSafeShell
def NBSafeShell(
path, extra_paths:NoneType=None, allow:NoneType=None, ok_dests:NoneType=None, cache_httpx2:bool=False,
cache_dir:NoneType=None, cache_domains:NoneType=None, execution_policy:NoneType=None
):Timeouts that do not trap future runs
A timed-out cell can make repeated verification painful. Timeout metadata records the source fingerprint that timed out, skips that exact source on later runs, and automatically clears the mark when the cell changes.
verify_notebook
def verify_notebook(
path, up2id:NoneType=None, timeout:int=30, safe:bool=True
):The public executor
exec_nb is the user-facing wrapper around the execution engine. It writes outputs back to the notebook, prints visible outputs when requested, and supports partial execution through up2id or chapter.
exec_nb
def exec_nb(
path:str, # Notebook path
dest:str | None=None, # Destination path; defaults to overwriting path
exc_stop:bool=False, # Stop on exceptions
up2id:int | str | None=None, # Execute first N cells, or through this cell id
chapter:str | None=None, # Execute through this chapter, inclusive
timeout:int=30, # Per-cell timeout in seconds; <=0 disables timeouts
show_output:bool=True, # Print saved cell outputs and errors after execution
verbose:bool=False, # Show stdout/stderr live while executing
safe:bool=True, # Use safepyrun instead of the legacy execnb shell
allow:str | None=None, # Comma-separated or literal list of trusted callables to allow
ok_dests:str | None=None, # Comma-separated or literal list of allowed write destinations
cache_httpx2:bool=False, # Return cached httpx2 responses instead of making live calls
cache_dir:str | None=None, # Directory containing cachy.jsonl; defaults to project root
cache_domains:str | None=None, # Comma-separated or literal list of cacheable domains
allow_new:bool=False, # Execute cells without prior user/nbskill execution approval
check_only:bool=False, # Run in memory without writing notebook outputs or metadata
):Execute a notebook with safe Python by default and local project imports available.
With fastcore 2, nbdev tests are treated as direct Python calls, so exec_nb(check_only=True) returns the notebook path while still leaving the file unchanged.
root = demo_path("03_execute_project")def _make_execute_project(name):
root = demo_path(name)
root.mkdir()
(root / "pyproject.toml").write_text("[project]\nname = 'local-demo'\n", encoding="utf-8")
pkg = root / "local_demo"
pkg.mkdir()
(pkg / "__init__.py").write_text("def meaning():\n return 42\n", encoding="utf-8")
nbs = root / "nbs"
nbs.mkdir()
return root, nbsroot, nbs = _make_execute_project("03_execute_project")
try:
path = nbs / "sample.ipynb"
write_nb(str(path), "%%code\nfrom local_demo import meaning\nprint(meaning())\nassert meaning() == 42", replace=True)
exec_nb(str(path), timeout=5, allow="local_demo.meaning")
text = "".join(output_text(output) for output in read_nb(path).cells[0].outputs)
assert "refusing to execute unapproved cell" in text
exec_nb(str(path), timeout=5, allow="local_demo.meaning", allow_new=True)
text = "".join(output_text(output) for output in read_nb(path).cells[0].outputs)
assert "42" in text
site_root = root / ".venv" / "lib" / f"python{sys.version_info.major}.{sys.version_info.minor}" / "site-packages"
site_root.mkdir(parents=True)
(site_root / "project_only_dep.py").write_text("def answer():\n return 'project env'\n", encoding="utf-8")
project_env = nbs / "project_env.ipynb"
write_nb(str(project_env), "%%code\nfrom project_only_dep import answer\nprint(answer())", replace=True)
exec_nb(str(project_env), timeout=5, allow="project_only_dep.answer", allow_new=True)
text = "".join(output_text(output) for output in read_nb(project_env).cells[0].outputs)
assert "project env" in text
finally:
remove_demo_path(root)root, nbs = _make_execute_project("03_execute_check")
try:
check = nbs / "check_only.ipynb"
write_nb(str(check), "%%code\nprint('check only')", replace=True)
before = check.read_text(encoding="utf-8")
exec_nb(str(check), timeout=5, allow_new=True, check_only=True)
assert check.read_text(encoding="utf-8") == before
cross = nbs / "cross.ipynb"
write_nb(str(cross), "%%code\nx = 2\n---\n%%code\nprint(x + 3)", replace=True)
exec_nb(str(cross), timeout=5, allow_new=True)
text = "".join(output_text(output) for output in read_nb(cross).cells[1].outputs)
assert "5" in text
skipped = nbs / "eval_false.ipynb"
write_nb(str(skipped), "%%code\n#| eval: false\nraise RuntimeError('should not run')", replace=True)
exec_nb(str(skipped), timeout=5, allow_new=True)
assert read_nb(skipped).cells[0].outputs == []
finally:
remove_demo_path(root)root, nbs = _make_execute_project("03_execute_safety")
try:
slow = nbs / "slow.ipynb"
write_nb(str(slow), "%%code\nimport time\ntime.sleep(2)", replace=True)
exec_nb(str(slow), timeout=1, allow_new=True, safe=False)
cell = read_nb(slow).cells[0]
assert cell.metadata["nbskill"][_TIMEOUT_HASH_KEY]
text = "".join(output_text(output) for output in cell.outputs)
assert "ran longer than 1s" in text
exec_nb(str(slow), timeout=1, allow_new=True, safe=False)
cell = read_nb(slow).cells[0]
text = "".join(output_text(output) for output in cell.outputs)
assert "skipped cell" in text
update_cell(str(slow), "print('fast now')", cell_id=cell.id)
exec_nb(str(slow), timeout=1, allow_new=True, safe=False)
cell = read_nb(slow).cells[0]
assert _TIMEOUT_HASH_KEY not in cell.metadata.get("nbskill", {})
text = "".join(output_text(output) for output in cell.outputs)
assert "fast now" in text
finally:
remove_demo_path(root)root, nbs = _make_execute_project("03_execute_blocks")
try:
blocked_cases = {
"path_write": "from pathlib import Path\nPath('bad.txt').write_text('bad')",
"open_write": "open('bad.txt', 'w').write('bad')",
"remove": "import os\nos.remove('missing.txt')",
"rmtree": "import shutil\nshutil.rmtree('missing')",
"subprocess": "import subprocess\nsubprocess.run(['true'])",
"magic": "%time 1 + 1",
"shell": "!echo unsafe",
}
for name, source in blocked_cases.items():
nb_path = nbs / f"{name}.ipynb"
write_nb(str(nb_path), f"%%code\n{source}", replace=True)
exec_nb(str(nb_path), timeout=5, allow_new=True)
text = "".join(output_text(output) for output in read_nb(nb_path).cells[0].outputs)
assert text
if name in {"magic", "shell"}: assert "blocks IPython" in text
finally:
remove_demo_path(root)root, nbs = _make_execute_project("03_execute_write")
try:
allowed_write = nbs / "allowed_write.ipynb"
ok_file = root / "allowed.txt"
write_nb(str(allowed_write), f"%%code\nfrom pathlib import Path\nPath({str(ok_file)!r}).write_text('ok')\nprint(Path({str(ok_file)!r}).read_text())", replace=True)
exec_nb(str(allowed_write), timeout=5, ok_dests=str(root), allow_new=True)
assert ok_file.read_text(encoding="utf-8") == "ok"
finally:
remove_demo_path(root)root, nbs = _make_execute_project("03_execute_httpx2")
try:
import httpx2
live = nbs / "live_httpx2.ipynb"
write_nb(str(live), "%%code\nimport httpx2\nhttpx2.get('https://api.openai.com/v1/test')", replace=True)
exec_nb(str(live), timeout=5, allow_new=True)
text = "".join(output_text(output) for output in read_nb(live).cells[0].outputs)
assert "blocked live httpx2 call" in text
cached = nbs / "cached_httpx2.ipynb"
url = "https://api.openai.com/v1/test"
request = httpx2.Request("GET", url)
key = _cachy_key(request)
(root / "cachy.jsonl").write_text(
json.dumps({"key": key, "response": "cached body", "headers": {"content-type": "text/plain"}, "status_code": 200}) + "\n",
encoding="utf-8",
)
write_nb(str(cached), f"%%code\nimport httpx2\nr = httpx2.get({url!r})\nprint(r.text)", replace=True)
exec_nb(str(cached), timeout=5, cache_httpx2=True, cache_dir=str(root), allow_new=True)
text = "".join(output_text(output) for output in read_nb(cached).cells[0].outputs)
assert "cached body" in text
finally:
remove_demo_path(root)root, nbs = _make_execute_project("03_execute_httpx2_miss")
try:
miss = nbs / "miss_httpx2.ipynb"
miss_url = "https://api.openai.com/v1/miss"
write_nb(str(miss), f"%%code\nimport httpx2\nhttpx2.get({miss_url!r})", replace=True)
exec_nb(str(miss), timeout=5, cache_httpx2=True, cache_dir=str(root), allow_new=True)
text = "".join(output_text(output) for output in read_nb(miss).cells[0].outputs)
assert "no cached httpx2 response" in text
finally:
remove_demo_path(root)warning_root = demo_path("03_execute_uncalled_warning")
try:
warning_root.mkdir()
warning_nbs = warning_root / "nbs"
warning_nbs.mkdir()
uncalled = warning_nbs / "uncalled.ipynb"
write_nb(str(uncalled), "%%code\ndef unused_demo():\n return 1\n---\n%%code\nprint('done')", replace=True)
result = exec_nb(str(uncalled), timeout=5, allow_new=True, show_output=False)
assert any("unused_demo" in warning for warning in result["warnings"])
finally:
remove_demo_path(warning_root)warning_root = demo_path("03_execute_called_warning")
try:
warning_root.mkdir()
warning_nbs = warning_root / "nbs"
warning_nbs.mkdir()
called = warning_nbs / "called.ipynb"
write_nb(str(called), "%%code\ndef used_demo():\n return 1\n---\n%%code\nused_demo()", replace=True)
out = StringIO()
with redirect_stdout(out):
exec_nb(str(called), timeout=5, allow_new=True, show_output=False)
assert "used_demo" not in out.getvalue()
finally:
remove_demo_path(warning_root)warning_root = demo_path("03_execute_decorated_warning")
try:
warning_root.mkdir()
warning_nbs = warning_root / "nbs"
warning_nbs.mkdir()
decorated = warning_nbs / "decorated.ipynb"
write_nb(str(decorated), "%%code\ndef _route(func):\n return func\n\n@_route\ndef api_handler():\n return 1", replace=True)
out = StringIO()
with redirect_stdout(out):
exec_nb(str(decorated), timeout=5, allow_new=True, show_output=False)
assert "api_handler" not in out.getvalue()
finally:
remove_demo_path(warning_root)Testing after edits
The write tools can ask for a notebook test immediately after changing a notebook. These helpers summarize saved error outputs so a failed write/test cycle reports the useful cell-level problem.