Coverage for src/zapy/utils/module.py: 98%
38 statements
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
1import importlib.util
2import types
3from pathlib import Path
4from typing import Any
6from zapy.templating.eval import exec_async
9async def load_python(module_path: str | Path) -> Any:
10 module_path = Path(module_path)
11 module_spec = importlib.util.spec_from_file_location(module_path.stem, module_path)
12 if module_spec is None or module_spec.loader is None:
13 err_msg = "module spec is none"
14 raise ValueError(err_msg)
15 module = importlib.util.module_from_spec(module_spec)
16 module_spec.loader.exec_module(module)
18 return module
21async def load_ipynb(module_path: str | Path, variables: dict[str, Any] | None = None) -> Any:
22 """from https://jupyter-notebook.readthedocs.io/en/latest/examples/Notebook/Importing%20Notebooks.html"""
23 from IPython import get_ipython
24 from IPython.core.interactiveshell import InteractiveShell
25 from nbformat import read
27 module_path = Path(module_path)
28 shell = InteractiveShell.instance()
29 fullname = module_path.stem
31 variables = variables or {}
33 # load the notebook object
34 with open(module_path, encoding="utf-8") as f:
35 nb = read(f, 4)
37 # create the module and add it to sys.modules
38 mod = types.ModuleType(fullname)
39 mod.__file__ = str(module_path)
40 mod.__dict__["get_ipython"] = get_ipython
41 # apply parameters
42 for k, v in variables.items():
43 mod.__dict__[k] = v
45 # extra work to ensure that magics that would affect the user_ns
46 # actually affect the notebook module's ns
47 save_user_ns = shell.user_ns
48 shell.user_ns = mod.__dict__
50 try:
51 for cell in nb.cells:
52 if cell.cell_type == "code": 52 ↛ 51line 52 didn't jump to line 51, because the condition on line 52 was never false
53 # transform the input to executable Python
54 code = shell.input_transformer_manager.transform_cell(cell.source)
55 # run the code in the module
56 await exec_async(code, mod.__dict__)
57 finally:
58 shell.user_ns = save_user_ns
59 return mod