From 0ed21e6789c00daef8c111387f70761fe6010220 Mon Sep 17 00:00:00 2001 From: Niklas Peng Date: Fri, 30 May 2025 19:49:53 +0200 Subject: [PATCH] =?UTF-8?q?src/execution=5Fengine.py=20gel=C3=B6scht?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/execution_engine.py | 119 ---------------------------------------- 1 file changed, 119 deletions(-) delete mode 100644 src/execution_engine.py diff --git a/src/execution_engine.py b/src/execution_engine.py deleted file mode 100644 index 7896ab3..0000000 --- a/src/execution_engine.py +++ /dev/null @@ -1,119 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -"""Code execution module for the codeeditor application.""" - -import io -import logging -import subprocess -import sys -import tempfile -import traceback -from typing import Dict, Tuple - -logger = logging.getLogger(__name__) - - -class ExecutionEngine: - """Class for executing code and capturing output.""" - - def __init__(self): - """Initialize the ExecutionEngine.""" - logger.info("ExecutionEngine initialized") - - def run_code(self, code: str) -> Dict[str, str]: - """Run Python code and capture the output. - - Args: - code: The Python code to execute. - - Returns: - A dictionary containing 'output', 'error', and 'status'. - """ - result = {"output": "", "error": "", "status": "success"} - - try: - # Create a temporary file to store the code - with tempfile.NamedTemporaryFile( - suffix=".py", delete=False, mode="w" - ) as temp_file: - temp_file.write(code) - temp_file_path = temp_file.name - - # Run the code in a subprocess - process = subprocess.Popen( - [sys.executable, temp_file_path], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) - - # Capture output and errors - stdout, stderr = process.communicate(timeout=10) # 10 second timeout - - result["output"] = stdout - - if stderr: - result["error"] = stderr - result["status"] = "error" - - logger.info(f"Code execution completed with status: {result['status']}") - - except subprocess.TimeoutExpired: - result["error"] = "Execution timed out after 10 seconds" - result["status"] = "timeout" - logger.warning("Code execution timed out") - - except Exception as e: - result["error"] = f"Error executing code: {str(e)}" - result["status"] = "error" - logger.error(f"Error executing code: {e}") - - return result - - def run_code_in_memory(self, code: str) -> Dict[str, str]: - """Run Python code in memory and capture the output. - - This method is safer for small code snippets as it doesn't write to disk. - - Args: - code: The Python code to execute. - - Returns: - A dictionary containing 'output', 'error', and 'status'. - """ - result = {"output": "", "error": "", "status": "success"} - - # Redirect stdout and stderr - old_stdout = sys.stdout - old_stderr = sys.stderr - redirected_output = io.StringIO() - redirected_error = io.StringIO() - sys.stdout = redirected_output - sys.stderr = redirected_error - - try: - # Execute the code - exec(code) - result["output"] = redirected_output.getvalue() - - error_output = redirected_error.getvalue() - if error_output: - result["error"] = error_output - result["status"] = "error" - - logger.info( - f"In-memory code execution completed with status: {result['status']}" - ) - - except Exception as e: - result["error"] = f"{str(e)}\n{traceback.format_exc()}" - result["status"] = "error" - logger.error(f"Error executing code in memory: {e}") - - finally: - # Restore stdout and stderr - sys.stdout = old_stdout - sys.stderr = old_stderr - - return result