360 lines
14 KiB
Python
360 lines
14 KiB
Python
from flask import Blueprint, render_template, redirect, url_for, flash, request, jsonify
|
|
from jinja2_fragments import render_block
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
from flask_login import current_user, login_required
|
|
from extensions import db, htmx, environment
|
|
from datetime import datetime, timezone, timedelta
|
|
import json
|
|
from services import create_http_function_view_model, create_http_functions_view_model
|
|
|
|
'''
|
|
CREATE TABLE http_function_versions (
|
|
id SERIAL PRIMARY KEY,
|
|
http_function_id INT NOT NULL,
|
|
script_content TEXT NOT NULL,
|
|
version_number INT NOT NULL,
|
|
versioned_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
|
|
CONSTRAINT fk_http_function_versions
|
|
FOREIGN KEY (http_function_id)
|
|
REFERENCES http_functions (id)
|
|
ON DELETE CASCADE
|
|
ON UPDATE CASCADE
|
|
);
|
|
|
|
CREATE OR REPLACE FUNCTION fn_http_functions_versioning()
|
|
RETURNS TRIGGER
|
|
LANGUAGE plpgsql
|
|
AS $$
|
|
DECLARE
|
|
next_version INT;
|
|
BEGIN
|
|
IF TG_OP = 'INSERT' THEN
|
|
INSERT INTO http_function_versions (http_function_id, script_content, version_number)
|
|
VALUES (NEW.id, NEW.script_content, 1);
|
|
|
|
UPDATE http_functions
|
|
SET version_number = 1
|
|
WHERE id = NEW.id;
|
|
|
|
RETURN NEW;
|
|
|
|
ELSIF TG_OP = 'UPDATE' THEN
|
|
IF NEW.script_content IS DISTINCT FROM OLD.script_content THEN
|
|
SELECT COALESCE(MAX(version_number), 0) + 1
|
|
INTO next_version
|
|
FROM http_function_versions
|
|
WHERE http_function_id = NEW.id;
|
|
|
|
INSERT INTO http_function_versions (http_function_id, script_content, version_number)
|
|
VALUES (NEW.id, NEW.script_content, next_version);
|
|
|
|
UPDATE http_functions
|
|
SET version_number = next_version
|
|
WHERE id = NEW.id;
|
|
END IF;
|
|
|
|
RETURN NEW;
|
|
END IF;
|
|
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
|
|
CREATE TRIGGER tr_http_functions_versioning
|
|
AFTER INSERT OR UPDATE
|
|
ON http_functions
|
|
FOR EACH ROW
|
|
EXECUTE PROCEDURE fn_http_functions_versioning();
|
|
'''
|
|
|
|
DEFAULT_SCRIPT = """async (req) => {
|
|
console.log(`Method:${req.method}`)
|
|
console.log(`Generating ${environment.lines} random lines...`)
|
|
let svgContent = "";
|
|
|
|
for (let i = 0; i < environment.lines; i++) {
|
|
console.log(i)
|
|
let pathD = "M2 " + Math.random() * 79;
|
|
let circles = "";
|
|
|
|
for (let x = 12; x <= 202; x += 10) {
|
|
let y = Math.random() * 79
|
|
pathD += ` L${x} ${y}`;
|
|
circles += `<circle cx="${x}" cy="${y}" r="1"></circle>`;
|
|
}
|
|
|
|
let pathColor = `rgb(${Math.random() * 255}, ${Math.random() * 255}, ${Math.random() * 255})`;
|
|
svgContent += `
|
|
<g style="fill: ${pathColor}; stroke: ${pathColor};">
|
|
<path d="${pathD}" fill="none" stroke="${pathColor}"></path>
|
|
${circles}
|
|
</g>`;
|
|
}
|
|
|
|
return HtmlResponse(`<svg viewBox="0 0 204 79" preserveAspectRatio="none">${svgContent}</svg>`);
|
|
}"""
|
|
|
|
DEFAULT_ENVIRONMENT = """{
|
|
"lines": 3
|
|
}"""
|
|
|
|
http = Blueprint('http', __name__)
|
|
|
|
@http.route("/", methods=["GET"])
|
|
@login_required
|
|
def dashboard_http_functions():
|
|
user_id = current_user.id
|
|
http_functions = db.get_http_functions_for_user(user_id)
|
|
http_functions = create_http_functions_view_model(http_functions)
|
|
if htmx:
|
|
return render_block(environment, "dashboard/http_functions/overview.html", "page", http_functions=http_functions)
|
|
return render_template("dashboard/http_functions/overview.html", http_functions=http_functions)
|
|
|
|
|
|
@http.route("/add_form", methods=["GET"])
|
|
@login_required
|
|
def get_http_function_add_form():
|
|
user_id = current_user.id
|
|
if htmx:
|
|
return render_block(environment, 'dashboard/http_functions/new.html', 'page', user_id=user_id, name='foo', script=DEFAULT_SCRIPT, environment_info=DEFAULT_ENVIRONMENT, is_public=False, log_request=True, log_response=False)
|
|
return render_template("dashboard/http_functions/new.html", user_id=user_id, name='foo', script=DEFAULT_SCRIPT, environment_info=DEFAULT_ENVIRONMENT, is_public=False, log_request=True, log_response=False)
|
|
|
|
|
|
@http.route("/create", methods=["POST"])
|
|
@login_required
|
|
def create_http_function():
|
|
try:
|
|
user_id = current_user.id
|
|
name = request.form.get('name')
|
|
script_content = request.form.get('script_content')
|
|
environment_info = request.form.get('environment_info')
|
|
is_public = request.form.get('is_public')
|
|
log_request = request.form.get('log_request')
|
|
log_response = request.form.get('log_response')
|
|
|
|
db.create_new_http_function(user_id, name, script_content, environment_info, is_public, log_request, log_response)
|
|
|
|
http_functions = db.get_http_functions_for_user(user_id)
|
|
http_functions = create_http_functions_view_model(http_functions)
|
|
return render_block(environment, "dashboard/http_functions/overview.html", "page", http_functions=http_functions), 200, {"HX-Push-Url": url_for('http.dashboard_http_functions')}
|
|
except Exception as e:
|
|
print(e)
|
|
return { "status": "error", "message": str(e) }
|
|
|
|
|
|
@http.route("/<int:function_id>/edit", methods=["POST"])
|
|
@login_required
|
|
def edit_http_function(function_id):
|
|
try:
|
|
user_id = current_user.id
|
|
name = request.json.get('name')
|
|
script_content = request.json.get('script_content')
|
|
environment_info = request.json.get('environment_info')
|
|
is_public = request.json.get('is_public')
|
|
log_request = request.json.get('log_request')
|
|
log_response = request.json.get('log_response')
|
|
|
|
updated_version = db.edit_http_function(user_id, function_id, name, script_content, environment_info, is_public, log_request, log_response)
|
|
|
|
return { "status": "success", "message": f'{name} updated' }
|
|
except Exception as e:
|
|
print(e)
|
|
return { "status": "error", "message": str(e) }
|
|
|
|
@http.route("/<int:function_id>/delete", methods=["DELETE"])
|
|
@login_required
|
|
def delete_http_function(function_id):
|
|
try:
|
|
user_id = current_user.id
|
|
db.delete_http_function(user_id, function_id)
|
|
|
|
http_functions = db.get_http_functions_for_user(user_id)
|
|
http_functions = create_http_functions_view_model(http_functions)
|
|
return render_block(environment, "dashboard/http_functions/overview.html", "page", http_functions=http_functions), 200, {"HX-Push-Url": url_for('http.dashboard_http_functions')}
|
|
except Exception as e:
|
|
return jsonify({"status": 'error', "message": str(e)}), 500
|
|
|
|
@http.route("/<int:function_id>/logs", methods=["GET"])
|
|
@login_required
|
|
def get_http_function_logs(function_id):
|
|
user_id = current_user.id
|
|
http_function = db.get_http_function_by_id(user_id, function_id)
|
|
if not http_function:
|
|
return jsonify({'error': 'Function not found'}), 404
|
|
name = http_function['name']
|
|
http_function_invocations = db.get_http_function_invocations(function_id)
|
|
if htmx:
|
|
return render_block(environment, 'dashboard/http_functions/logs.html', 'page', user_id=user_id, function_id=function_id, name=name, http_function_invocations=http_function_invocations)
|
|
return render_template("dashboard/http_functions/logs.html", user_id=user_id, name=name, function_id=function_id, http_function_invocations=http_function_invocations)
|
|
|
|
@http.route("/<int:function_id>/client", methods=["GET"])
|
|
@login_required
|
|
def client(function_id):
|
|
user_id = current_user.id
|
|
http_function = db.get_http_function_by_id(user_id, function_id)
|
|
if not http_function:
|
|
return jsonify({'error': 'Function not found'}), 404
|
|
|
|
http_function = create_http_function_view_model(http_function)
|
|
if htmx:
|
|
return render_block(environment, 'dashboard/http_functions/client.html', 'page', function_id=function_id, **http_function)
|
|
return render_template("dashboard/http_functions/client.html", function_id=function_id, **http_function)
|
|
|
|
@http.route("/<int:function_id>/history", methods=["GET"])
|
|
@login_required
|
|
def get_http_function_history(function_id):
|
|
user_id = current_user.id
|
|
http_function = db.get_http_function_by_id(user_id, function_id)
|
|
if not http_function:
|
|
return jsonify({'error': 'Function not found'}), 404
|
|
|
|
name = http_function['name']
|
|
version_number = http_function['version_number']
|
|
original_script = http_function['script_content'] if version_number == 1 else None
|
|
|
|
http_function_history = []
|
|
if version_number > 1:
|
|
raw_history = db.get_http_function_history(function_id)
|
|
|
|
for i in range(len(raw_history) - 1):
|
|
pre_version = raw_history[i + 1]
|
|
post_version = raw_history[i]
|
|
|
|
http_function_history.append({
|
|
'pre': pre_version['script_content'],
|
|
'post': post_version['script_content'],
|
|
'version_id': post_version['version_id'],
|
|
'version_number': post_version['version_number'],
|
|
'updated_at': post_version['updated_at']
|
|
})
|
|
|
|
if raw_history:
|
|
original_script = raw_history[-1]['script_content']
|
|
|
|
if htmx:
|
|
return render_block(environment, 'dashboard/http_functions/history.html', 'page', user_id=user_id, function_id=function_id, name=name, http_function=http_function, http_function_history=http_function_history, original_script=original_script)
|
|
return render_template("dashboard/http_functions/history.html", user_id=user_id, name=name, function_id=function_id, http_function=http_function, http_function_history=http_function_history, original_script=original_script)
|
|
|
|
@http.route("/editor/<int:function_id>", methods=["GET"])
|
|
@login_required
|
|
def http_function_editor(function_id):
|
|
user_id = current_user.id
|
|
http_function = db.get_http_function_by_id(user_id, function_id)
|
|
if not http_function:
|
|
return jsonify({'error': 'Function not found'}), 404
|
|
|
|
# Create a view model with all necessary data for the editor
|
|
editor_data = {
|
|
'id': http_function['id'],
|
|
'name': http_function['name'],
|
|
'script_content': http_function['script_content'],
|
|
'environment_info': json.dumps(http_function['environment_info'], indent=2),
|
|
'is_public': http_function['is_public'],
|
|
'log_request': http_function['log_request'],
|
|
'log_response': http_function['log_response'],
|
|
'version_number': http_function['version_number'],
|
|
'user_id': user_id,
|
|
'function_id': function_id,
|
|
# Add new URLs for navigation
|
|
'cancel_url': url_for('http.dashboard_http_functions'),
|
|
'edit_url': url_for('http.http_function_editor', function_id=function_id),
|
|
}
|
|
|
|
if htmx:
|
|
return render_block(environment, "dashboard/http_functions/editor.html", "page", **editor_data)
|
|
|
|
return render_template("dashboard/http_functions/editor.html", **editor_data)
|
|
@http.route("/api", methods=["POST"])
|
|
@login_required
|
|
def api_create_http_function():
|
|
try:
|
|
user_id = current_user.id
|
|
data = request.get_json()
|
|
name = data.get('name')
|
|
script_content = data.get('script_content')
|
|
environment_info = data.get('environment_info')
|
|
is_public = data.get('is_public')
|
|
log_request = data.get('log_request')
|
|
log_response = data.get('log_response')
|
|
|
|
# Check if function with same name already exists for this user
|
|
existing_function = db.get_http_function(user_id, name)
|
|
if existing_function:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": f"A function with the name '{name}' already exists"
|
|
}), 400
|
|
|
|
http_function = db.create_new_http_function(
|
|
user_id,
|
|
name,
|
|
script_content,
|
|
environment_info,
|
|
is_public,
|
|
log_request,
|
|
log_response
|
|
)
|
|
|
|
return jsonify({
|
|
"status": "success",
|
|
"message": f'{name} created',
|
|
"function": http_function
|
|
})
|
|
except Exception as e:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": str(e)
|
|
}), 400
|
|
|
|
@http.route("/api/<int:function_id>", methods=["POST"])
|
|
@login_required
|
|
def api_update_http_function(function_id):
|
|
try:
|
|
user_id = current_user.id
|
|
data = request.get_json()
|
|
name = data.get('name')
|
|
script_content = data.get('script_content')
|
|
environment_info = data.get('environment_info')
|
|
is_public = data.get('is_public')
|
|
log_request = data.get('log_request')
|
|
log_response = data.get('log_response')
|
|
|
|
updated_function = db.edit_http_function(
|
|
user_id,
|
|
function_id,
|
|
name,
|
|
script_content,
|
|
environment_info,
|
|
is_public,
|
|
log_request,
|
|
log_response
|
|
)
|
|
|
|
return jsonify({
|
|
"status": "success",
|
|
"message": f'{name} updated',
|
|
"function": updated_function
|
|
})
|
|
except Exception as e:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": str(e)
|
|
}), 400
|
|
|
|
@http.route("/api/<int:function_id>", methods=["DELETE"])
|
|
@login_required
|
|
def api_delete_http_function(function_id):
|
|
try:
|
|
user_id = current_user.id
|
|
db.delete_http_function(user_id, function_id)
|
|
|
|
return jsonify({
|
|
"status": "success",
|
|
"message": "Function deleted successfully"
|
|
})
|
|
except Exception as e:
|
|
return jsonify({
|
|
"status": "error",
|
|
"message": str(e)
|
|
}), 400 |