本文共 10098 字，大约阅读时间需要 33 分钟。

参考文档：

目录结构：

(vpp_agent_venv3.7) [root@localhost source]# tree .├── app│   ├── __init__.py│   ├── views│   │   ├── ik_agent_view.py│   │   ├── ik_echarts_view.py│   │   ├── ik_file_view.py│   │   ├── ik_test_view.py│   │   ├── ik_user_view.py│   │   ├── ik_vpp_view.py│   │   └── __init__.py├── release│   └── service│       ├── ptt_agent.service│       └── vpp.service├── release_note.txt├── requirements.txt├── templates│   ├── components.html│   ├── index2.html│   ├── index.html│   ├── internal_server_error.html│   ├── macro│   ├── nb_components.html│   ├── nb_jupyter_globe.html│   ├── nb_jupyter_lab.html│   ├── nb_jupyter_lab_tab.html│   ├── nb_jupyter_notebook.html│   ├── nb_jupyter_notebook_tab.html│   ├── nb_nteract.html│   ├── simple_chart.html│   ├── simple_globe.html│   ├── simple_page.html│   └── simple_tab.html├── gunicorn_config.py└── wsgi.py

bp+模板渲染+wsgi：

定义蓝图

# rom flask import Flaskfrom jinja2 import Markupimport randomfrom pyecharts import options as optsfrom pyecharts.charts import Barfrom flask import Blueprintfrom flask import request, render_templatefrom app.ik_global import ik_globaldebug = ik_global.Logger.debugbp = Blueprint('echarts', __name__)def bar_base() -> Bar:    c = (        Bar()            .add_xaxis(["衬衫", "羊毛衫", "雪纺衫", "裤子", "高跟鞋", "袜子"])            .add_yaxis("商家A", [5, 20, 36, 10, 75, 90])            .add_yaxis("商家B", [15, 25, 16, 55, 48, 8])            .set_global_opts(title_opts=opts.TitleOpts(title="Bar-基本示例", subtitle="我是副标题"))    )    return cdef bar_base2(title="shen", subtitle="xianjie") -> Bar:    c = (        Bar()            .add_xaxis(["衬衫", "羊毛衫", "雪纺衫", "裤子", "高跟鞋", "袜子"])            .add_yaxis("商家A", [random.randint(10, 100) for _ in range(6)])            .add_yaxis("商家B", [random.randint(10, 100) for _ in range(6)])            .set_global_opts(title_opts=opts.TitleOpts(title=title, subtitle=subtitle))    )    return c""" @bp.route("/")def index():    c = bar_base()    return Markup(c.render_embed())"""@bp.route("/")def index():    args = request.args.to_dict()    debug("args--> {}".format(args))    return render_template("index.html")  @bp.route("/barChart")def get_bar_chart():    args = request.args.to_dict()    name = args.get("name")    subtitle = args.get("subtitle")    c = bar_base2(name, subtitle)    return c.dump_options_with_quotes()

蓝图注册

app/__init__.py，在注册蓝图时，通过template_folder指定蓝图echarts的template文件加载路径（template_folder="../../templates"），这里echarts使用相对路径寻找template，也可以使用绝对路径。

#!/usr/bin/env python# -*- coding:utf-8 -*-# File: ik_agent.py# import sysfrom flask import Flask, jsonify# from flask_jwt import JWTfrom config import configimport multiprocessing as mpfrom app.ik_global import ik_global, vpp_sig_registerfrom werkzeug.contrib.cache import SimpleCache# from app.vpp.vpp import Pttoolfrom app.comm import ik_comm# from app.comm.agent_process import AgentProcessfrom app.log import logfrom app.ik_global.ik_global import MP_LISTfrom app.utils.threads import DaemonThreadTimerfrom app.comm.ik_thread_work import monitor_vpp_instance_statusfrom app.utils.ik_util import show_memory_infofrom app.utils.auth import authenticate, identityfrom app.utils.threads import set_intervalfrom app.ikdb.exts import dbfrom flask_jwt import JWTfrom werkzeug.contrib.profiler import ProfilerMiddlewarefrom werkzeug.wsgi import DispatcherMiddlewarefrom collections import OrderedDict# from app.views import json_pagedef create_app(config_name):    app = Flask(__name__, template_folder="/srv/web_develop/vpp_agent_dev/agent/source/templates")    app.config.from_object(config[config_name])    app.config.setdefault('JWT_SECRET_KEY', 'secret')    app.debug = True    # db init    # 方式一    db.init_app(app=app)  # 初始化连接数据库    # 方式二    # from app.ikdb.exts import connect_db    # init log    # log.init_log(app)    ik_global.Logger = log.create_logger()    # ik_global.set_printi()    # ik_global.printi("ssss")    # mongodb init    from app.ikdb.model import models_mongo    ik_global.APP_INSTANCE = app    ik_global.CACHE = SimpleCache()    ik_global.USER_CACHE = SimpleCache()    # blueprint regester    from app.views import ik_echarts_view    app.register_blueprint(            ik_echarts_view.bp, url_prefix="/echarts", template_folder="../../templates")    ik_global.Logger.info("性能测试工具节点--agent启动成功!")    ik_global.Logger.info("版本号: {}".format(ik_global.__version__))    ik_global.Logger.info("agent: {} port: {}".format(        config[config_name].LISTEN_ADDR, config[config_name].VPP_AGENT_LISTEN_PORT))    ik_global.Logger.info(show_memory_info("after app started"))    return app

设置gunicorn_config.py

# -*- coding: utf-8 -*-import osimport multiprocessing# from app.ik_global.ik_global import VPP_INSTANCEipaddr = os.getenv("LISTEN_ADDR")port = os.getenv("VPP_AGENT_LISTEN_PORT")# bind = '10.2.6.117:8082' #your app ip address and portbind = "{}:{}".format(ipaddr, port)  # your app ip address and portbacklog = 2048## Worker processes# workers = 1workers = multiprocessing.cpu_count() * 2 + 1worker_class = 'sync'# worker_class = 'gevent'worker_connections = 500# timeout = 120timeout = 0keepalive = 2##   spew - Install a trace function that spews every line of Python#       that is executed when running the server. This is the#       nuclear option.##       True or False#spew = False## Server mechanics#daemon = Falsepidfile = Noneumask = 0user = Nonegroup = Nonetmp_upload_dir = None##   Logging##   logfile - The path to a log file to write to.##       A path string. "-" means log to stdout.##   loglevel - The granularity of log output##       A string of "debug", "info", "warning", "error", "critical"## logconfig = 'logging.conf'# accesslog = 'log/gunicorn_access.log'# errorlog = 'log/gunicorn_error.log'# loglevel = 'warning'# access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s"'## Process namingproc_name = 'ik_pttool'def post_fork(server, worker):    server.log.info("Worker spawned (pid: %s)", worker.pid)def pre_fork(server, worker):    passdef pre_exec(server):    server.log.info("Forked child, re-executing.")def when_ready(server):    server.log.info("Server is ready. Spawning workers")def worker_int(worker):    worker.log.info("worker received INT or QUIT signal")    # get traceback info    import threading    import sys    import traceback    id2name = dict([(th.ident, th.name) for th in threading.enumerate()])    code = []    for threadId, stack in sys._current_frames().items():        code.append("\n# Thread: %s(%d)" % (id2name.get(threadId, ""),                                            threadId))        for filename, lineno, name, line in traceback.extract_stack(stack):            code.append('File: "%s", line %d, in %s' % (filename,                                                        lineno, name))            if line:                code.append("  %s" % (line.strip()))    worker.log.debug("\n".join(code))def worker_abort(worker):    worker.log.info("worker received SIGABRT signal")

配置wsgi.py

from app.ikdb.exts import dbimport osimport flask_loginfrom app.utils.ik_util import show_memory_info# print(show_memory_info("inital"))from flask import abort, request, _app_ctx_stack, jsonify, url_forfrom app import create_appfrom app.ik_global import ik_globalfrom app.ikdb.model.models import Userfrom flask_sqlalchemy import get_debug_queries# import profile# def get_debug_queries():#     return getattr(_app_ctx_stack.top, 'sqlalchemy_queries', [])app = vpp_agent = create_app("production")info = ik_global.Logger.info# profile.run('create_app("production")')login_manager = flask_login.LoginManager()login_manager.init_app(app)# 测试使用，重置数据库表# with app.app_context():#     # db.drop_all()#     db.create_all()@app.before_first_requestdef setup():    ik_global.Logger.info("before_first_request: {}".format(request.url))@app.before_requestdef before_request():    ik_global.Logger.info("before_request: {}".format(request.url))#     if not ik_global.VPP_INSTANCE:#         ik_global.Logger.info("service or device restarting!")#         ik_global.Logger.info("vpp is None")#         abort(503)@app.after_requestdef after_request(response):    ik_global.Logger.info("after_request: {}".format(request.url))    for query in get_debug_queries():        if query.duration >= app.config["DATABASE_QUERY_TIMEOUT"]:            ik_global.Logger.info("Context:{}\n SLOW query: {}\n Params: {}\n Duration: {} ms\n".format(                query.context, query.statement, query.parameters, query.duration*1000))    # some stuff here    ik_global.Logger.info(        "os.getpid():{}\n os.getppid: {}\n".format(os.getpid(), os.getppid()))    response.headers['Access-Control-Allow-Origin'] = '*'    response.headers['Access-Control-Allow-Headers'] = 'Content-Type'    return response@app.teardown_appcontextdef teardown(exc=None):    ik_global.Logger.info("teardown_appcontext: ")""" @app.context_processordef template_extras():    ik_global.Logger.info("context_processor:") """""" @app.errorhandler(404)def page_not_found(error):    ik_global.Logger.info("errorhandler: {}".format(error))    return jsonify({"code": 404, "result": "page_not_found"})"""@app.errorhandler(401)def unauthorized(error):    ik_global.Logger.info("errorhandler: {}".format(error))    return jsonify({"code": 401, "result": "unauthorized"})@flask_login.user_logged_in.connect_via(app)  # 信号订阅user_logged_in.connect(fn)def _track_logins(sender, user, **extra):    print("_track_logins")    user.login_count += 1    user.last_login_ip = request.remote_addr    result = User.update_one(user.id, **user.serialize)    ik_global.Logger.info("_track_logins < login_count:{} last_login_ip: {} >".format(        user.login_count, user.last_login_ip))@login_manager.user_loaderdef user_loader(id):    print("user_loader")    user = User.one(id=id)    ik_global.Logger.info("user_loader: {}".format(user.name))    return user@app.template_filter('capitalize')def reverse_filter(s):    ik_global.Logger.info("template_filter:")

模板文件：templates/index.html

模板中通过setInterval定时请求接口的数据，然后将数据渲染到html

效果：数据其实是动态变化的

转载地址：http://hsfld.baihongyu.com/