PythonInterpreter that runs code in a sandboxed environment using Deno and Pyodide.
Prerequisites:
- Deno (https://docs.deno.com/runtime/getting_started/installation/).
Example Usage:
code_string = "print('Hello'); 1 + 2"
with PythonInterpreter() as interp:
output = interp(code_string) # If final statement is non-None, prints the numeric result, else prints captured output
Parameters:
| Name |
Type |
Description |
Default |
deno_command
|
Optional[List[str]]
|
command list to launch Deno.
|
None
|
enable_read_paths
|
Optional[List[Union[PathLike, str]]]
|
Files or directories to allow reading from in the sandbox.
|
None
|
enable_write_paths
|
Optional[List[Union[PathLike, str]]]
|
Files or directories to allow writing to in the sandbox.
|
None
|
enable_env_vars
|
Optional[List[str]]
|
Environment variable names to allow in the sandbox.
|
None
|
enable_network_access
|
Optional[List[str]]
|
Domains or IPs to allow network access in the sandbox.
|
None
|
sync_files
|
bool
|
If set, syncs changes within the sandbox back to original files after execution.
|
True
|
Source code in dspy/primitives/python_interpreter.py
| def __init__(
self,
deno_command: Optional[List[str]] = None,
enable_read_paths: Optional[List[Union[PathLike, str]]] = None,
enable_write_paths: Optional[List[Union[PathLike, str]]] = None,
enable_env_vars: Optional[List[str]] = None,
enable_network_access: Optional[List[str]] = None,
sync_files: bool = True,
) -> None:
"""
Args:
deno_command: command list to launch Deno.
enable_read_paths: Files or directories to allow reading from in the sandbox.
enable_write_paths: Files or directories to allow writing to in the sandbox.
enable_env_vars: Environment variable names to allow in the sandbox.
enable_network_access: Domains or IPs to allow network access in the sandbox.
sync_files: If set, syncs changes within the sandbox back to original files after execution.
"""
if isinstance(deno_command, dict):
deno_command = None # no-op, just a guard in case someone passes a dict
self.enable_read_paths = enable_read_paths or []
self.enable_write_paths = enable_write_paths or []
self.enable_env_vars = enable_env_vars or []
self.enable_network_access = enable_network_access or []
self.sync_files = sync_files
# TODO later on add enable_run (--allow-run) by proxying subprocess.run through Deno.run() to fix 'emscripten does not support processes' error
if deno_command:
self.deno_command = list(deno_command)
else:
args = ["deno", "run", "--allow-read"]
self._env_arg = ""
if self.enable_env_vars:
user_vars = [str(v).strip() for v in self.enable_env_vars]
args.append("--allow-env=" + ",".join(user_vars))
self._env_arg = ",".join(user_vars)
if self.enable_network_access:
args.append(f"--allow-net={','.join(str(x) for x in self.enable_network_access)}")
if self.enable_write_paths:
args.append(f"--allow-write={','.join(str(x) for x in self.enable_write_paths)}")
args.append(self._get_runner_path())
# For runner.js to load in env vars
if self._env_arg:
args.append(self._env_arg)
self.deno_command = args
self.deno_process = None
self._mounted_files = False
|
Functions
__call__(code: str, variables: Optional[Dict[str, Any]] = None) -> Any
Source code in dspy/primitives/python_interpreter.py
| def __call__(
self,
code: str,
variables: Optional[Dict[str, Any]] = None,
) -> Any:
return self.execute(code, variables)
|
execute(code: str, variables: Optional[Dict[str, Any]] = None) -> Any
Source code in dspy/primitives/python_interpreter.py
| def execute(
self,
code: str,
variables: Optional[Dict[str, Any]] = None,
) -> Any:
variables = variables or {}
code = self._inject_variables(code, variables)
self._ensure_deno_process()
self._mount_files()
# Send the code as JSON
input_data = json.dumps({"code": code})
try:
self.deno_process.stdin.write(input_data + "\n")
self.deno_process.stdin.flush()
except BrokenPipeError:
# If the process died, restart and try again once
self._ensure_deno_process()
self.deno_process.stdin.write(input_data + "\n")
self.deno_process.stdin.flush()
# Read one JSON line from stdout
output_line = self.deno_process.stdout.readline().strip()
if not output_line:
# Possibly the subprocess died or gave no output
err_output = self.deno_process.stderr.read()
raise InterpreterError(f"No output from Deno subprocess. Stderr: {err_output}")
# Parse that line as JSON
try:
result = json.loads(output_line)
except json.JSONDecodeError:
# If not valid JSON, just return raw text
result = {"output": output_line}
# If we have an error, determine if it's a SyntaxError or other error using error.errorType.
if "error" in result:
error_msg = result["error"]
error_type = result.get("errorType", "Sandbox Error")
if error_type == "FinalAnswer":
# The `FinalAnswer` trick to receive output from the sandbox interpreter,
# just simply replace the output with the arguments.
result["output"] = result.get("errorArgs", None)
elif error_type == "SyntaxError":
raise SyntaxError(f"Invalid Python syntax. message: {error_msg}")
else:
raise InterpreterError(f"{error_type}: {result.get('errorArgs') or error_msg}")
# If there's no error or got `FinalAnswer`, return the "output" field
self._sync_files()
return result.get("output", None)
|
shutdown() -> None
Source code in dspy/primitives/python_interpreter.py
| def shutdown(self) -> None:
if self.deno_process and self.deno_process.poll() is None:
shutdown_message = json.dumps({"shutdown": True}) + "\n"
self.deno_process.stdin.write(shutdown_message)
self.deno_process.stdin.flush()
self.deno_process.stdin.close()
self.deno_process.wait()
self.deno_process = None
|