src/execution_engine.py gelöscht
This commit is contained in:
parent
a84501b690
commit
0ed21e6789
@ -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
|
Loading…
x
Reference in New Issue
Block a user