当前位置：首页 > news >正文

数据库flask访问

news 2025/10/18 7:05:36

执行节点

import urllib.request
import jsondef main(sql: str) -> dict:api_url = "http://192.168.0.108:35003/execute_query"headers = {"Content-Type": "application/json"}data = json.dumps({"sql": sql}).encode("utf-8")try:req = urllib.request.Request(api_url, data=data, headers=headers, method="POST")with urllib.request.urlopen(req) as response:res_data = json.loads(response.read().decode("utf-8"))# 转成 JSON 字符串返回return {"result": json.dumps(res_data)}except Exception as e:# 异常信息也返回字符串return {"result": f"请求接口失败: {str(e)}"}

app.py

运行方法：

yum install -y python3

pip3 install flask pymysql

python3 app.py

firewall-cmd --add-port=35003/tcp --permanent

firewall-cmd --reload

from flask import Flask, request, jsonify
import pymysqlapp = Flask(__name__)@app.route("/execute_query", methods=['GET', 'POST'])
def execute_query():data = request.get_json()sql = data.get("sql", "").strip()if not sql:return jsonify({"error": "SQL语句不能为空"}), 400if not sql.lower().startswith("select"):return jsonify({"error": "只允许执行SELECT语句"}), 400try:# 建立数据库连接conn = pymysql.connect(host="192.168.0.108",  # 你的数据库IPuser="root",password="123456",database="mydb",cursorclass=pymysql.cursors.DictCursor)with conn.cursor() as cursor:cursor.execute(sql)result = cursor.fetchall()conn.close()return jsonify(result)except pymysql.Error as e:return jsonify({"error": f"数据库错误: {str(e)}"}), 500except Exception as e:return jsonify({"error": f"未知错误: {str(e)}"}), 500if __name__ == "__main__":app.run(host="0.0.0.0", port=35003)

查看全文

http://www.dtcms.com/a/494981.html