这篇文章主要介绍了python flask数据可视化怎么实现的相关知识,内容详细易懂,操作简单快捷,具有一定借鉴价值,相信大家阅读完这篇python flask数据可视化怎么实现文章都会有所收获,下面我们一起来看看吧。
flask server文件
# -*- coding: utf-8 -*- # 作者: lang1.xia # 创建时间: 2022-08-25 03:38:26 # 修改时间: 2022-08-25 03:38:26 from flask import Flask from jinja2 import Environment, FileSystemLoader from markupsafe import Markup from pyecharts.globals import CurrentConfig # echarts 外部样式调用 CurrentConfig.GLOBAL_ENV = Environment(loader=FileSystemLoader("./templates")) CurrentConfig.ONLINE_HOST = 'http://127.0.0.1/assets/' from pyecharts.options import ComponentTitleOpts from pyecharts.components import Table # 外部数据库方法调用 from data_storage import one_data # 自定义本地html模板 app = Flask(__name__, template_folder="templates") def bar_base() -> Table: # 查询数据库信息形成大列表 sql = "select * from ains_mysql_base" res = one_data(sql) data_rows = [] if res["code"] == 200 and res["data"]: for i in res["data"]: disk_io_info = ''.join(list(i[8])) disk_io_tmp = "" for key_d in eval(disk_io_info): key = list(key_d.keys()) val = list(key_d.values()) new_str = key[0] + ": " + str(round(val[0],4)) + " " disk_io_tmp += new_str new_i = list(i[3:7]) + list(i[9:-1]) new_i.append(disk_io_tmp) disk_info = ''.join(list(i[7])) disk_tmp = "" for key_d in eval(disk_info): key = list(key_d.keys()) val = list(key_d.values()) new_str = key[0] + ": " + str(round(val[0],4)) + " " disk_tmp += new_str new_i.append(disk_tmp) data_rows.append(new_i) print(data_rows) # 定义表头 headers = [ "IP地址", "CPU使用率", "CPU五分钟负载","内存使用率", "innodb行锁", "连接数","磁盘IO", "磁盘"] rows = data_rows # 添加标题、表数据、表样式 c = ( Table() .add(headers, rows, attributes={"style": "margin:0% auto;font-size: 28px;text-align: left,width:100px", "class": "fl-table"}) .set_global_opts(title_opts=ComponentTitleOpts(title="数据库巡检", title_style={"style": "font-size: 28px; font-weight:bold;text-align: center"})) ) return c @app.route("/") def index(): # 调用函数、返回到前端 c = bar_base() return Markup(c.render_embed()) if __name__ == "__main__": app.run(host="0.0.0.0")
数据库SQL执行封装文件
# -*- coding: utf-8 -*- # 作者: lang1.xia # 创建时间: 2022-08-22 07:06:53 # 修改时间: 2022-08-22 07:06:53 # 未使用 import pymysql # 基础连接信息 def conndb(): conn = pymysql.connect(host="IP地址", user="账号", passwd="密码", database="数据库名", port="数据库端口") cur = conn.cursor() return conn, cur # 关闭连接 def closedb(conn, cur): cur.close() conn.close() # executemany方法封装 def batch_data(sql): conn, cur = conndb() try: cur.executemany(sql) result = cur.fetchall() except Exception as e : return {"code": 400, "message": e} else: conn.commit() closedb(conn=conn, cur=cur) return {"code": 200, "message": "true", "data": result} # execute方法封装 def one_data(sql): conn, cur = conndb() try: cur.execute(sql) result = cur.fetchall() except Exception as e : return {"code": 400, "message": e} else: conn.commit() closedb(conn=conn, cur=cur) return {"code": 200, "message": "true", "data": result}
运行flask服务
python server.py
前端访问
浏览器访问http://127.0.0.1:5000