Execute Python code. Saves the script to a file if save_to_file is True and a parent and filename are provided. Returns: { stdout, stderr, return_code, execution_time, file_path?, error? }
(
ctx: Context,
code: str,
timeout: int = 30,
*,
parent: Optional[str] = None,
filename: Optional[str] = None,
save_to_file: bool = False,
)
| 39 | |
| 40 | |
| 41 | async def execute_python_code( |
| 42 | ctx: Context, |
| 43 | code: str, |
| 44 | timeout: int = 30, |
| 45 | *, |
| 46 | parent: Optional[str] = None, |
| 47 | filename: Optional[str] = None, |
| 48 | save_to_file: bool = False, |
| 49 | ) -> dict: |
| 50 | """ |
| 51 | Execute Python code. Saves the script to a file if save_to_file is True and a parent and filename are provided. |
| 52 | |
| 53 | Returns: { stdout, stderr, return_code, execution_time, file_path?, error? } |
| 54 | """ |
| 55 | sid = getattr(ctx, "session_id", None) |
| 56 | if not sid: |
| 57 | return _err("session_id_REQUIRED", message="ctx.session_id is missing") |
| 58 | |
| 59 | rt = MANAGER.get(sid) |
| 60 | env = rt.minimal_env() |
| 61 | tout = _clamp_timeout(timeout) |
| 62 | |
| 63 | try: |
| 64 | base_dir = ( |
| 65 | (rt.root / parent.lstrip("/")) if parent else (rt.root / "tmp" / "code") |
| 66 | ).resolve() |
| 67 | base_dir.mkdir(parents=True, exist_ok=True) |
| 68 | except Exception as e: |
| 69 | return _err(type(e).__name__, message=str(e)) |
| 70 | |
| 71 | clean_code = textwrap.dedent(code or "").strip() + "\n" |
| 72 | |
| 73 | temp_file_path: Optional[str] = None |
| 74 | persisted = False |
| 75 | result = { |
| 76 | "stdout": "", |
| 77 | "stderr": "", |
| 78 | "return_code": -1, |
| 79 | "execution_time": 0.0, |
| 80 | } |
| 81 | |
| 82 | try: |
| 83 | with tempfile.NamedTemporaryFile( |
| 84 | mode="w", suffix=".py", dir=base_dir, delete=False |
| 85 | ) as tf: |
| 86 | tf.write(clean_code) |
| 87 | temp_file_path = tf.name |
| 88 | # tempfile.NamedTemporaryFile creates files with 0600 (mkstemp); |
| 89 | # relax to 0644 so the host process can read them for backups. |
| 90 | os.chmod(temp_file_path, 0o644) |
| 91 | |
| 92 | if save_to_file: |
| 93 | final_name = _sanitize_filename( |
| 94 | filename or f"code_{Path(temp_file_path).name}" |
| 95 | ) |
| 96 | final_path = Path(base_dir) / final_name |
| 97 | os.replace(temp_file_path, final_path) |
| 98 | temp_file_path = str(final_path) |
nothing calls this directly
no test coverage detected