

<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

## Executing 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.

``` python
exec_nb("nbs/03_execute.ipynb", up2id="some-cell-id", timeout=10, show_output=True)
```

### Production contract

Notebook execution is production core.
[`exec_nb`](https://MorsCerta-crypto.github.io/nbskill/execute.html#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.

``` python
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_nb
```

``` python
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`](https://MorsCerta-crypto.github.io/nbskill/execute.html#_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.

------------------------------------------------------------------------

<a
href="https://github.com/MorsCerta-crypto/nbskill/blob/main/nbskill/execute.py#L700"
target="_blank" style="float:right; font-size:smaller">source</a>

### NBSafeShell

``` python
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.

------------------------------------------------------------------------

<a
href="https://github.com/MorsCerta-crypto/nbskill/blob/main/nbskill/execute.py#L980"
target="_blank" style="float:right; font-size:smaller">source</a>

### verify_notebook

``` python
def verify_notebook(
    path, up2id:NoneType=None, timeout:int=30, safe:bool=True
):
```

### The public executor

[`exec_nb`](https://MorsCerta-crypto.github.io/nbskill/execute.html#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`.

------------------------------------------------------------------------

<a
href="https://github.com/MorsCerta-crypto/nbskill/blob/main/nbskill/execute.py#L1079"
target="_blank" style="float:right; font-size:smaller">source</a>

### exec_nb

``` python
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.

``` python
root = demo_path("03_execute_project")
```

``` python
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, nbs
```

``` python
root, 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)
```

``` python
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)
```

``` python
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)
```

``` python
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)
```

``` python
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)
```

``` python
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)
```

``` python
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)
```

``` python
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)
```

``` python
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)
```

``` python
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.

------------------------------------------------------------------------

<a
href="https://github.com/MorsCerta-crypto/nbskill/blob/main/nbskill/execute.py#L1143"
target="_blank" style="float:right; font-size:smaller">source</a>

### run_notebook_test

``` python
def run_notebook_test(
    path, timeout:int=30
):
```