资讯专栏INFORMATION COLUMN

基于 flask-socketio 的 CRUD 操作初探

K_B_Z / 1840人阅读

摘要:理解协议协议只能通过客户端发起请求来与客户端进行通讯这是一个缺陷。与协议有着良好的兼容性。以下的表格内容显示数据局里的内容,每秒局部刷新一次表格内容。欢迎大侠能够给我的项目提出修改意见,先行感谢源码下载参考基于的操作教程阮一峰

Flask 作为一个全栈架构,如果你只会 python,而不懂 javascript 的前端知识,似乎是无法支撑起你的 web 梦想的,比如,一个简单的页面 局部刷新 功能,你就需要用到 ajax 的知识,当然,你还可以使用 HTML5 的新特性 —— websocket功能,好在 flask 还提供了一个 flask-socketio 插件,本文我们就探讨一下这个 flask-scoketio插件的用法。
理解 websocket 协议

HTTP 协议只能通过客户端发起请求来与客户端进行通讯 —— 这是一个缺陷。

通过websocket 协议,服务器可以主动向客户端推送信息,客户端也可以主动向服务器发送信息,是真正的双向平等对话,属于服务器推送技术的一种。

websocket 协议特性

建立在 TCP 协议之上,服务器端的实现比较容易。

与 HTTP 协议有着良好的兼容性。默认端口也是80和443,并且握手阶段采用 HTTP 协议,因此握手时不容易屏蔽,能通过各种 HTTP 代理服务器。

数据格式比较轻量,性能开销小,通信高效。

可以发送文本,也可以发送二进制数据。

没有同源限制,客户端可以与任意服务器通信。

协议标识符是ws(如果加密,则为wss),服务器网址就是 URL。

使用 flask-socketio 安装插件
pip install flask-socketio
项目结构

本文是在 《基于 flask 的 CRUD 操作》 的基础上增加了 webscoket 的功能,使用的是 init_app() 的形式加载 flask-socketio 插件,和网上的大多数教程稍有不同

flask-wtf-crud/
|-- env/
    |-- 
|-- app/ <项目的模块名称>
    |-- crud/ <前端蓝图>
        |-- __init__.py
        |-- views.py <路由和视图函数文件>
        |-- forms.py <表单类文件, wtforms插件必须项>
        |-- templates 
            |-- static <静态文件夹>
                |-- js 
                    |-- crud.js # 异步请求的程序主要在此添加
    |-- XXXXXX/ <其它蓝图>
    |-- __init__.py
    |-- models.py <数据库模型文件>
|-- migrations/ <数据库表关系文件夹,Flask-Migrate迁移数据库时使用>
|-- config.py <项目的配置文件>
|-- manage.py <用于启动程序以及其它程序任务>
将 flask-socketio 引入项目 修改 manage.py 内容
# -*- coding:utf-8 -*-
__author__ = "东方鹗"
__blog__ = u"http://www.os373.cn"

import os
from app import create_app, db, socketio
from app.models import User
from flask_script import Manager, Shell
from flask_migrate import Migrate, MigrateCommand


app = create_app(os.getenv("FLASK_CONFIG") or "default")
manager = Manager(app=app)
migrate = Migrate(app=app, db=db)

def make_shell_context():
    return dict(app=app, db=db, User=User)


manager.add_command("shell", Shell(make_context=make_shell_context))
manager.add_command("db", MigrateCommand)
manager.add_command("run", socketio.run(app=app, host="0.0.0.0", port=5001)) # 新加入的内容


if __name__ == "__main__":
    manager.run()
 修改 app/__init__.py 内容
# -*- coding:utf-8 -*-
__author__ = "东方鹗"
__blog__ = u"http://www.os373.cn"

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from config import config
from flask_socketio import SocketIO # 新加入的内容
db = SQLAlchemy()

async_mode = None
socketio = SocketIO()


def create_app(config_name):
    """ 使用工厂函数初始化程序实例"""
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app=app)

    db.init_app(app=app)

    socketio.init_app(app=app, async_mode=async_mode) # 新加入的内容

    # 注册蓝本crud
    from .crud import crud as crud_blueprint
    app.register_blueprint(crud_blueprint, url_prefix="/crud")

    return app
当前蓝图的 views.py
# -*- coding:utf-8 -*-
__author__ = "东方鹗"
__blog__ = u"http://www.os373.cn"

from flask import render_template, redirect, request, current_app, url_for, flash, json
from . import crud
from ..models import User
from .forms import AddUserForm, DeleteUserForm, EditUserForm
from ..import db
from threading import Lock
from app import socketio # 新加入的内容
from flask_socketio import emit # 新加入的内容

# 新加入的内容-开始
thread = None
thread_lock = Lock()

def background_thread(users_to_json):
    """Example of how to send server generated events to clients."""
    while True:
        socketio.sleep(5)  每五秒发送一次

        socketio.emit("user_response", {"data": users_to_json}, namespace="/websocket/user_refresh")
# 新加入的内容-结束

@crud.route("/", methods=["GET", "POST"])
def index():

    return render_template("index.html")


@crud.route("/websocket", methods=["GET", "POST"])
def websocket():
    add_user_form = AddUserForm(prefix="add_user")
    delete_user_form = DeleteUserForm(prefix="delete_user")
    if add_user_form.validate_on_submit():
        if add_user_form.role.data == u"True":
            role = True
        else:
            role = False
        if add_user_form.status.data == u"True":
            status = True
        else:
            status = False
        u = User(username=add_user_form.username.data.strip(), email=add_user_form.email.data.strip(),
                 role=role, status=status)
        db.session.add(u)
        flash({"success": u"添加用户<%s>成功!" % add_user_form.username.data.strip()})
    if delete_user_form.validate_on_submit():
        u = User.query.get_or_404(int(delete_user_form.user_id.data.strip()))
        db.session.delete(u)
        flash({"success": u"删除用户<%s>成功!" % u.username})

    users = User.query.all()

    return render_template("websocket.html", users=users, addUserForm=add_user_form, deleteUserForm=delete_user_form)


@crud.route("/websocket-edit/", methods=["GET", "POST"])
def user_edit(user_id):
    user = User.query.get_or_404(user_id)
    edit_user_form = EditUserForm(prefix="edit_user", obj=user)
    if edit_user_form.validate_on_submit():
        user.username = edit_user_form.username.data.strip()
        user.email = edit_user_form.email.data.strip()
        if edit_user_form.role.data == u"True":
            user.role = True
        else:
            user.role = False
        if edit_user_form.status.data == u"True":
            user.status = True
        else:
            user.status = False
        flash({"success": u"用户资料已修改成功!"})
        return redirect(url_for(".basic"))

    return render_template("edit_websocket.html", editUserForm=edit_user_form, user=user)

# 新加入的内容-开始
@socketio.on("connect", namespace="/websocket/user_refresh")
def connect():
    """ 服务端自动发送通信请求 """
    global thread
    with thread_lock:
        users = User.query.all()
        users_to_json = [user.to_json() for user in users]

        if thread is None:
            thread = socketio.start_background_task(background_thread, (users_to_json, ))
    emit("server_response", {"data": "试图连接客户端!"})


@socketio.on("connect_event", namespace="/websocket/user_refresh")
def refresh_message(message):
    """ 服务端接受客户端发送的通信请求 """

    emit("server_response", {"data": message["data"]})
# 新加入的内容-结束

---------- 以上内容是后端的内容,以下内容是将是前段的内容 ----------

crud.js 内容
$(document).ready(function () {
    namespace="/websocket/user_refresh";
    var socket = io.connect(location.protocol + "//" + document.domain + ":" + location.port + namespace);
    $("#url_show").text("websocket URL: " + location.protocol + "//" + document.domain + ":" + location.port + namespace);

    socket.on("connect", function() { // 发送到服务器的通信内容
        socket.emit("connect_event", {data: "我已连接上服务端!"});
    });

    socket.on("server_response", function(msg) {
         显示接受到的通信内容,包括服务器端直接发送的内容和反馈给客户端的内容
        $("#log").append("
" + $("
").text("接收 : " + msg.data).html()); }); socket.on("user_response", function(msg) { //console.log(eval(msg.data[0])); //$("#users_show").append("
" + $("
").text("接收 : " + msg.data).html()); var tbody = ""; var obj = eval(msg.data[0]); $.each(obj, function (n, value) { var role = ""; if (value.role===true){ role = "管理员"; }else { role = "一般用户"; } var status = ""; if (value.status===true){ status = "正常"; }else { status = "注销"; } edit_url = " 修改"; delete_url = "删除"; var trs = ""; trs += "" + (n+1) + "" + value.username + "" + value.email + "" + role + "" + status + "" + edit_url + " | " + delete_url +""; tbody += trs; }) $("#users_show").empty(); $("#users_show").append(tbody); }); });
显示结果

每次打开网页,会显示服务端发送的内容——“试图连接客户端!”,其后,客户端返回给服务端——“我已连接上服务端!”,而后又被服务端返回给客户端显示。

以下的表格内容显示数据局里的内容,每 5 秒局部刷新一次表格内容。

服务器后端 log 日志内容如下:

总结

由于 flask 架构具有上下文的限制,在数据库里 增加删改 内容的时候,表格的内容没有变化——尽管局部已经进行了刷新。要想显示变化后的数据库内容,必须得重新启动一下 flask 服务。

就整体的部署来说,在 flask 项目里添加 websocket 协议,显得项目较重,实现一个局部刷新的功能还是用 ajax 比较简单。

欢迎大侠能够给我的项目提出修改意见,先行感谢!!!

源码下载

参考

基于 flask 的 CRUD 操作

WebSocket 教程 —— 阮一峰

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/44491.html

相关文章

  • 利用K8S技术栈打造个人私有云(连载之:K8S资源控制)

    摘要:将用户命令通过接口传送给,从而进行资源的增删改等操作。要使用编写应用程序,当下大多语言都可以很方便地去实现请求来操作的接口从而控制和查询资源,但本文主要是利用已有的客户端来更加优雅地实现的资源控制。 showImg(https://segmentfault.com/img/remote/1460000013517345); 【利用K8S技术栈打造个人私有云系列文章目录】 利用K8S...

    Reducto 评论0 收藏0
  • 利用K8S技术栈打造个人私有云(连载之:K8S资源控制)

    摘要:将用户命令通过接口传送给,从而进行资源的增删改等操作。要使用编写应用程序,当下大多语言都可以很方便地去实现请求来操作的接口从而控制和查询资源,但本文主要是利用已有的客户端来更加优雅地实现的资源控制。 showImg(https://segmentfault.com/img/remote/1460000013517345); 【利用K8S技术栈打造个人私有云系列文章目录】 利用K8S...

    Render 评论0 收藏0
  • 基于websocketcelery任务状态监控

    摘要:目的曾经想向前台实时返回任务的状态监控,也查看了很多博客,但是好多也没能如愿,因此基于网上已有的博客已经自己的尝试,写了一个小的,实现前台实时获取后台传输的任务状态。实现仿照其他例子实现了一个简单的后台任务监控。 1. 目的曾经想向前台实时返回Celery任务的状态监控,也查看了很多博客,但是好多也没能如愿,因此基于网上已有的博客已经自己的尝试,写了一个小的demo,实现前台实时获取后...

    microelec 评论0 收藏0
  • angular开发中问题记录--启动过程初探

    摘要:然而代码的最终执行结果表明,内的代码运行应该是先于里面的代码。项目中请求服务端异步获取数据的接口参考文档中几种的区别源码阅读启动过程 公司一些管理后台的前端页面,使用的是angular开发的,得益于angular的双向绑定和模块化controller使得构建pc端的CRUD应用简单了不少。angular有很多比较难理解的概念,上手起来没有vue简单,不过对着模板项目、看看tutoria...

    G9YH 评论0 收藏0
  • Babylon-AST初探-实战

    摘要:生成属性这一步,我们要先提取原函数中的的对象。所以这里我们还是主要使用来访问节点获取第一级的,也就是函数体将合并的写法用生成生成生成插入到原函数下方删除原函数程序输出将中的属性提升一级这里遍历中的属性没有再采用,因为这里结构是固定的。   经过之前的三篇文章介绍,AST的CRUD都已经完成。下面主要通过vue转小程序过程中需要用到的部分关键技术来实战。 下面的例子的核心代码依然是最简单...

    godiscoder 评论0 收藏0

发表评论

0条评论

K_B_Z

|高级讲师

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<