File size: 3,699 Bytes
0208d9d 0e20d87 0208d9d 0e20d87 0208d9d d165403 0208d9d d165403 0208d9d d165403 0208d9d d165403 0208d9d d165403 0208d9d d165403 9304b86 d165403 9304b86 0208d9d 0e20d87 d165403 0208d9d 9304b86 0e20d87 0208d9d 9304b86 0208d9d d165403 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | import subprocess
import sys
import uuid
from flask import Flask, request, jsonify
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
@app.route('/')
def home():
return jsonify({"status": "ok", "message": "API Running (Isolated Sessions Enabled)"})
def run_command(command, input_text=None, timeout=15):
"""Helper to run shell commands safely"""
try:
run_process = subprocess.run(
command,
input=input_text,
capture_output=True,
text=True,
timeout=timeout
)
output = run_process.stdout
if run_process.stderr:
output += "\n--- Error Log ---\n" + run_process.stderr
return output
except subprocess.TimeoutExpired:
return "Error: Execution timed out (15s limit)."
except Exception as e:
return f"System Error: {str(e)}"
@app.route('/api/run-python', methods=['POST'])
def run_python():
code = request.json.get('code', '')
user_input = request.json.get('input', '')
if not code: return jsonify({'output': 'No code provided'}), 400
cmd = [sys.executable, "-c", code]
output = run_command(cmd, input_text=user_input)
return jsonify({'output': output})
@app.route('/api/run-sql', methods=['POST'])
def run_sql():
data = request.json
sql_code = data.get('code', '')
db_type = data.get('type', 'sqlite').lower()
if not sql_code: return jsonify({'output': 'No SQL code provided'}), 400
output = ""
session_id = uuid.uuid4().hex[:8] # Random 8-char ID for isolation
try:
# 1. SQLITE (Already Isolated by :memory:)
if db_type == 'sqlite':
cmd = ["sqlite3", ":memory:"]
output = run_command(cmd, input_text=sql_code)
# 2. POSTGRESQL (Schema Isolation)
elif db_type in ['postgres', 'postgresql']:
schema_name = f"sess_{session_id}"
# A. Create Temp Schema
setup_cmd = ["psql", "-h", "localhost", "-U", "playground", "-d", "playground", "-c", f"CREATE SCHEMA {schema_name};"]
run_command(setup_cmd)
# B. Run User Code (Wrapped in search_path to use the schema)
# We modify the SQL to force it into the new schema
isolated_sql = f"SET search_path TO {schema_name};\n{sql_code}"
main_cmd = ["psql", "-h", "localhost", "-U", "playground", "-d", "playground"]
output = run_command(main_cmd, input_text=isolated_sql)
# C. Cleanup (Drop Schema)
cleanup_cmd = ["psql", "-h", "localhost", "-U", "playground", "-d", "playground", "-c", f"DROP SCHEMA {schema_name} CASCADE;"]
run_command(cleanup_cmd)
# 3. MYSQL (Database Isolation)
elif db_type == 'mysql':
db_name = f"sess_{session_id}"
# A. Create Temp Database
setup_cmd = ["mysql", "-h", "localhost", "-u", "playground", "-e", f"CREATE DATABASE {db_name};"]
run_command(setup_cmd)
# B. Run User Code (Against the new DB)
main_cmd = ["mysql", "-h", "localhost", "-u", "playground", "-D", db_name, "-t"]
output = run_command(main_cmd, input_text=sql_code)
# C. Cleanup (Drop Database)
cleanup_cmd = ["mysql", "-h", "localhost", "-u", "playground", "-e", f"DROP DATABASE {db_name};"]
run_command(cleanup_cmd)
else:
return jsonify({'output': 'Unsupported database type.'}), 400
except Exception as e:
output = f"Server Error: {str(e)}"
return jsonify({'output': output})
if __name__ == '__main__':
app.run(host="0.0.0.0", port=7860) |