backend.py

import os
import json
import traceback
from flask import Flask, request, jsonify, Response, stream_with_context, session
from flask_cors import CORS

from dotenv import load_dotenv

import io
from contextlib import redirect_stdout, redirect_stderr

# Initialization Code

from datetime import datetime
import numpy as np
import gc
import pathsim, pathsim_chem

print(f"PathSim {pathsim.__version__} loaded successfully")

STREAMING_STEP_EXPR = "_step_streaming_gen()"
NAMESPACE_LIFETIME = 60* 60 * 1000 # A namespace will persist for a maximum of 1 hour before deletion
_clean_globals = set(globals().keys())

'''
The Flask web server would not be initialized simultaneously with the SvelteKit website since the latter is statically generated,
rather there would be some type of deployment of this application such that it could receive requests from 
"https://view.pathsim.org" (which I think is already encapsualted by the "*" in the CORS.resources.options parameter)
'''

load_dotenv()

server_namespaces = {}

app = Flask(__name__, static_folder="../static", static_url_path="")

# app.secret_key = os.getenv("SECRET_KEY")
# app.config["SECRET_KEY"] = os.getenv("SECRET_KEY")

# app.config["SESSION_PERMANENT"] = False
# app.config["SESSION_TYPE"] = 'filesystem'
app.config.update(
    SECRET_KEY=os.getenv("SECRET_KEY"), # Required for session
    SESSION_COOKIE_SAMESITE='None',
    SESSION_COOKIE_SECURE=True,
    SESSION_COOKIE_HTTPONLY=True,
)
# app.config["SESSION_SERIALIZATION_FORMAT"] = 'json'
# app.config["SESSION_CACHELIB"] = FileSystemCache(threshold=500, cache_dir="/sessions"),

# Session(app)f

if os.getenv("FLASK_ENV") == "production":
    CORS(app,
         resources={
             r"/*": {
                 "origins": ["*"],
                 "methods": ["GET", "POST", "PUT", "DELETE", "OPTIONS"],
                 "allow_headers": ["Content-Type", "Authorization"]
             }
         }, supports_credentials=True)
else:
    print("We are not in production...")
    CORS(
        app, 
        resources={
            r"/*": {""
                "origins": ["http://localhost:5173", "http://localhost:3000"]
            }
        },
        supports_credentials=True
    )

@app.route("/initialize", methods=["GET"])
def initalize():
    session_id = None

    if "id" in session:
        session_id = session["id"]
        app.logger.info("We already have a session ID it is...")
        app.logger.info(session_id)
    else:
        app.logger.info("Making a session id...")
        session_id = str(os.urandom(12))
        app.logger.info("Made the id: ")
        app.logger.info(session_id)
        session["id"] = session_id

    server_namespaces[session_id] = {
        "namespace": {},
        "lifetime": NAMESPACE_LIFETIME,
        "created": datetime.now()
    }

    try:

        return jsonify({
            "success": True,
            "id": session_id
        })
    except Exception as e:
        return jsonify({
            "success": False,
            "error": e
        }), 400

@app.route("/idCheck", methods=["GET"])
def idCheck():
    session_id = None
    if "id" in session:
        session_id = session["id"]

    return jsonify({ "success": True, "id": session_id })

@app.route("/namespaceCheck", methods=["GET"])
def namespaceCheck():
    namespace = {}
    session_id = None
    if "id" in session:
        app.logger.info("The id exists...")
        session_id = session["id"]
        if session_id in server_namespaces:
            app.logger.info("Found an associated namespace...")
            namespace = server_namespaces[session["id"]]["namespace"]
    keys = ""
    if isinstance(namespace, dict):
        for k in server_namespaces.keys():
            keys += k + ", "

    return jsonify({ "success": True, "namespace_keys": keys})
# Execute Python route copied from the previous repository
@app.route("/execute-code", methods=["POST"])
def execute_code():
    """Execute Python code and returns nothing."""

    try:
        data = request.json
        code = data.get("code", "")

        if not code.strip():
            return jsonify({"success": False, "error": "No code provided"}), 400

        # Capture stdout and stderr
        stdout_capture = io.StringIO()
        stderr_capture = io.StringIO()

        user_namespace = {}
        app.logger.info("Session: ", session)
        app.logger.info("Session ID: ", session["id"])
        if "id" in session:
            user_namespace = server_namespaces[session["id"]]["namespace"]

        try:
            with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
                exec(code, user_namespace)

            app.logger.info("User Namespace: ", user_namespace)

            if "id" in session:
                server_namespaces[session["id"]]["namespace"] = user_namespace

            # Capture any output
            output = stdout_capture.getvalue()
            error_output = stderr_capture.getvalue()

            if error_output:
                return jsonify({"success": False, "error": error_output})

            return jsonify(
                {
                    "success": True,
                    "output": output,
                }
            )

        except SyntaxError as e:
            return jsonify({"success": False, "error": f"Syntax Error: {str(e)}"}), 400
        except Exception as e:
            return jsonify({"success": False, "error": f"Runtime Error: {str(e)}"}), 400

    except Exception as e:
        return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500

@app.route("/evaluate-expression", methods=["POST"])
def evaluate_expression():
    "Evaluates Python expression and returns result"
    try:
        data = request.json
        expr = data.get("expr")
        
        if not expr.strip():
            return jsonify({"success": False, "error": "No Python expression provided"}), 400

        stdout_capture = io.StringIO()
        stderr_capture = io.StringIO()

        user_namespace = {}
        if "id" in session:
            user_namespace = server_namespaces[session["id"]]["namespace"]

        try:
            result = ""
            with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
                result = eval(expr, user_namespace)

            app.logger.info("User Namespace: ", user_namespace)

            if "id" in session:
                server_namespaces[session["id"]]["namespace"] = user_namespace

            # Capture any output
            output = stdout_capture.getvalue()
            error_output = stderr_capture.getvalue()

            if error_output:
                return jsonify({"success": False, "error": error_output})
            
            return jsonify(
                {
                    "success": True,
                    "result": result,
                    "output": output
                }
            )

        except SyntaxError as e:
            return jsonify({"success": False, "error": f"Syntax Error: {str(e)}"}), 400
        except Exception as e:
            return jsonify({"success": False, "error": f"Runtime Error: {str(e)}"}), 400 
    except Exception as e:
        return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500


@app.route("/traceback", methods=["GET"])
def check_traceback():
    try:
        traceback_text = traceback.format_exc()
        return jsonify({"success": True, "traceback": traceback_text})
    except Exception as e:
        return jsonify({"success": False, "error": f"Server-side error: {e}"})

@app.route("/streamData", methods=["POST", "GET"])
def stream_data():
    def generate(expr):

        # Capture stdout and stderror
        stdout_capture = io.StringIO()
        stderr_capture = io.StringIO()

        user_namespace = {}
        if "id" in session:
            user_namespace = server_namespaces[session["id"]]["namespace"]

        isDone = False
        
        while not isDone:

            result = " "
            with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
                result = eval(expr, user_namespace)

            app.logger.info("User Namespace: ", user_namespace)

            if "id" in session:
                server_namespaces[session["id"]]["namespace"] = user_namespace

            # Capture any output
            output = stdout_capture.getvalue()
            error_output = stderr_capture.getvalue()

            if error_output:
                return jsonify({"success": False, "error": error_output})
            
            # Directly responding with a Flask Response object (as jsonify(...) does) doesn't work
            # so we need to use the json.dumps(...) function to return a string so that it can pass into
            # stream_with_context(...)

            yield json.dumps(
                {
                    "success": True,
                    "result": result,
                    "output": output
                }
            )

            if result["done"]:
                isDone = True
    
    try:
        method = request.method
        
        expr = STREAMING_STEP_EXPR

        if method == "POST":
            data = request.json
            expr = data.get("expr")

        try:
            return Response(stream_with_context(generate(expr)), content_type='application/json')
        
        except SyntaxError as e:
            return jsonify({"success": False, "error": f"Syntax Error: {str(e)}"}), 400
        except Exception as e:
            return jsonify({"success": False, "error": f"Runtime Error: {str(e)}"}), 400   


    except Exception as e:
        return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500


# Global error handler to ensure all errors return JSON
@app.errorhandler(Exception)
def handle_exception(e):
    """Global exception handler to ensure JSON responses."""
    import traceback
    from werkzeug.exceptions import HTTPException

    error_details = traceback.format_exc()
    print(f"Unhandled exception: {error_details}")

    # For HTTP exceptions, return a cleaner response
    if isinstance(e, HTTPException):
        return jsonify(
            {"success": False, "error": f"{e.name}: {e.description}"}
        ), e.code

    # For all other exceptions, return a generic JSON error
    return jsonify({"success": False, "error": f"Internal server error: {str(e)}"}), 500


if __name__ == "__main__":
    port = int(os.getenv("PORT", 8000))
    print("Hello there...our port is: ", port)
    print("Application Configuration: ", app.config)
    app.run(host="0.0.0.0", port=port, debug=os.getenv("FLASK_ENV") != "production")