perf 优化demo服务器

This commit is contained in:
kyj@bowong.ai 2025-07-30 18:22:58 +08:00
parent 4435a00dec
commit 591989e826
1 changed files with 121 additions and 71 deletions

View File

@ -1,108 +1,158 @@
from http.server import HTTPServer, BaseHTTPRequestHandler import sqlite3
import json import json
import threading from http.server import HTTPServer, BaseHTTPRequestHandler
import urllib import urllib.parse
# "数据库" 和线程锁保持不变 # --- 配置 ---
WORKFLOWS_DB = {} DATABASE_FILE = "workflows.sqlite"
db_lock = threading.Lock() PORT = 8000
class SimpleAPIHandler(BaseHTTPRequestHandler): # --- 数据库初始化 ---
def init_db():
"""初始化数据库,如果表不存在则创建它"""
with sqlite3.connect(DATABASE_FILE) as conn:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS workflows (
name TEXT PRIMARY KEY,
workflow_json TEXT NOT NULL
)
""")
conn.commit()
print(f"数据库 '{DATABASE_FILE}' 已准备就绪。")
# --- API 处理器 ---
class PersistentAPIHandler(BaseHTTPRequestHandler):
def _send_cors_headers(self): def _send_cors_headers(self):
"""发送CORS头部"""
self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'GET, POST, DELETE, OPTIONS') # 添加 DELETE self.send_header('Access-Control-Allow-Methods', 'GET, POST, DELETE, OPTIONS')
self.send_header("Access-Control-Allow-Headers", "X-Requested-With, Content-Type") self.send_header("Access-Control-Allow-Headers", "X-Requested-With, Content-Type")
def do_OPTIONS(self): def do_OPTIONS(self):
"""处理预检请求"""
self.send_response(200, "ok") self.send_response(200, "ok")
self._send_cors_headers() self._send_cors_headers()
self.end_headers() self.end_headers()
def do_GET(self): def do_GET(self):
# ... (do_GET 保持不变) ... """处理获取所有工作流的请求"""
if self.path == '/api/workflow': if self.path == '/api/workflow':
self.send_response(200)
self.send_header('Content-type', 'application/json')
self._send_cors_headers()
self.end_headers()
with db_lock:
response_data = list(WORKFLOWS_DB.values())
self.wfile.write(json.dumps(response_data).encode())
else:
self.send_response(404)
self.end_headers()
self.wfile.write(b'Not Found')
def do_POST(self):
# ... (do_POST 保持不变) ...
if self.path == '/api/workflow':
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
try: try:
data = json.loads(post_data) with sqlite3.connect(DATABASE_FILE) as conn:
name = data.get('name') conn.row_factory = sqlite3.Row # 让查询结果可以像字典一样访问
workflow_data = data.get('workflow') cursor = conn.cursor()
if not name or not workflow_data: cursor.execute("SELECT name, workflow_json FROM workflows")
raise ValueError("Missing 'name' or 'workflow' in request body") rows = cursor.fetchall()
with db_lock:
WORKFLOWS_DB[name] = {"name": name, "workflow": workflow_data} # 将数据库行转换为前端期望的JSON格式
print(f"--- Received/Updated workflow: {name} ---") workflows_list = [
{"name": row["name"], "workflow": json.loads(row["workflow_json"])}
for row in rows
]
self.send_response(200) self.send_response(200)
self.send_header('Content-type', 'application/json') self.send_header('Content-type', 'application/json')
self._send_cors_headers() self._send_cors_headers()
self.end_headers() self.end_headers()
self.wfile.write(json.dumps({"status": "received", "name": name}).encode()) self.wfile.write(json.dumps(workflows_list).encode())
except Exception as e:
self.send_response(400); except Exception as e:
self.end_headers(); self._send_error(f"获取工作流时出错: {e}")
self.wfile.write(f'Bad Request: {e}'.encode()) else:
else: self._send_error("路径未找到", 404)
self.send_response(404);
self.end_headers(); def do_POST(self):
self.wfile.write(b'Not Found') """处理发布/更新工作流的请求"""
if self.path == '/api/workflow':
try:
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
data = json.loads(post_data)
name = data.get('name')
workflow_data = data.get('workflow')
if not name or not workflow_data:
return self._send_error("请求体中缺少 'name''workflow'", 400)
workflow_json_str = json.dumps(workflow_data)
with sqlite3.connect(DATABASE_FILE) as conn:
cursor = conn.cursor()
# 使用 INSERT OR REPLACE 来处理创建和更新,非常方便
cursor.execute(
"INSERT OR REPLACE INTO workflows (name, workflow_json) VALUES (?, ?)",
(name, workflow_json_str)
)
conn.commit()
print(f"--- 已保存/更新工作流: {name} ---")
self.send_response(200)
self.send_header('Content-type', 'application/json')
self._send_cors_headers()
self.end_headers()
self.wfile.write(json.dumps({"status": "success", "name": name}).encode())
except Exception as e:
self._send_error(f"保存工作流时出错: {e}")
else:
self._send_error("路径未找到", 404)
# --- [核心新增] 添加 do_DELETE 方法 ---
def do_DELETE(self): def do_DELETE(self):
"""处理删除指定工作流的请求""" """处理删除指定工作流的请求"""
# 我们期望的路径格式: /api/workflow/工作流名称[版本号]
parts = self.path.split('/api/workflow/') parts = self.path.split('/api/workflow/')
if len(parts) == 2 and parts[1]: if len(parts) == 2 and parts[1]:
# URL解码工作流名称因为它可能包含空格等特殊字符
workflow_name_to_delete = urllib.parse.unquote(parts[1]) workflow_name_to_delete = urllib.parse.unquote(parts[1])
try:
with sqlite3.connect(DATABASE_FILE) as conn:
cursor = conn.cursor()
cursor.execute("DELETE FROM workflows WHERE name = ?", (workflow_name_to_delete,))
conn.commit()
with db_lock: # 检查是否真的有行被删除了
if workflow_name_to_delete in WORKFLOWS_DB: if cursor.rowcount > 0:
del WORKFLOWS_DB[workflow_name_to_delete] print(f"--- 已删除工作流: {workflow_name_to_delete} ---")
print(f"--- Deleted workflow: {workflow_name_to_delete} ---") self.send_response(200)
self.send_response(200) self.send_header('Content-type', 'application/json')
self.send_header('Content-type', 'application/json') self._send_cors_headers()
self._send_cors_headers() self.end_headers()
self.end_headers() self.wfile.write(json.dumps({"status": "deleted", "name": workflow_name_to_delete}).encode())
self.wfile.write(json.dumps({"status": "deleted", "name": workflow_name_to_delete}).encode()) else:
else: self._send_error("工作流未找到,无法删除", 404)
# 如果找不到要删除的工作流
self.send_response(404) except Exception as e:
self._send_cors_headers() self._send_error(f"删除工作流时出错: {e}")
self.end_headers()
self.wfile.write(json.dumps({"status": "error", "message": "Workflow not found"}).encode())
else: else:
self.send_response(400) self._send_error("无效的删除路径", 400)
self._send_cors_headers()
self.end_headers() def _send_error(self, message, code=500):
self.wfile.write(b'Bad Request: Invalid delete path') self.send_response(code)
self.send_header('Content-type', 'application/json')
self._send_cors_headers()
self.end_headers()
self.wfile.write(json.dumps({"status": "error", "message": message}).encode())
def run(server_class=HTTPServer, handler_class=SimpleAPIHandler, port=8000): # --- 主执行函数 ---
def run(server_class=HTTPServer, handler_class=PersistentAPIHandler, port=PORT):
# 1. 初始化数据库
init_db()
# 2. 启动服务器
server_address = ('', port) server_address = ('', port)
httpd = server_class(server_address, handler_class) httpd = server_class(server_address, handler_class)
print(f"测试API服务器已在 http://localhost:{port} 上启动...") print(f"\n持久化API服务器已在 http://localhost:{port} 上启动...")
print(" - POST /api/workflow: 发布新工作流") print(f"数据库文件: ./{DATABASE_FILE}")
print(" - GET /api/workflow: 获取所有工作流") print(" - POST /api/workflow -> 发布/更新工作流")
print(" - DELETE /api/workflow/<name>: 删除指定工作流") print(" - GET /api/workflow -> 获取所有工作流")
print(" - DELETE /api/workflow/<name> -> 删除指定工作流")
print("\n按 Ctrl+C 停止服务器。")
httpd.serve_forever() httpd.serve_forever()
if __name__ == '__main__': if __name__ == '__main__':
run() run()