jasonqiao36

jasonqiao36 查看完整档案

填写现居城市  |  填写毕业院校  |  填写所在公司/组织填写个人主网站
编辑
_ | |__ _ _ __ _ | '_ \| | | |/ _` | | |_) | |_| | (_| | |_.__/ \__,_|\__, | |___/ 个人简介什么都没有

个人动态

jasonqiao36 赞了文章 · 2017-12-16

基于 flask-socketio 的 CRUD 操作初探

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

理解 websocket 协议

  • HTTP 协议只能通过客户端发起请求来与客户端进行通讯 —— 这是一个缺陷。
  • 通过websocket 协议,服务器可以主动向客户端推送信息,客户端也可以主动向服务器发送信息,是真正的双向平等对话,属于服务器推送技术的一种。

websocket 协议特性

  1. 建立在 TCP 协议之上,服务器端的实现比较容易。
  2. 与 HTTP 协议有着良好的兼容性。默认端口也是80和443,并且握手阶段采用 HTTP 协议,因此握手时不容易屏蔽,能通过各种 HTTP 代理服务器。
  3. 数据格式比较轻量,性能开销小,通信高效。
  4. 可以发送文本,也可以发送二进制数据。
  5. 没有同源限制,客户端可以与任意服务器通信。
  6. 协议标识符是ws(如果加密,则为wss),服务器网址就是 URL。

使用 flask-socketio

安装插件

pip install flask-socketio

项目结构

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

flask-wtf-crud/
|-- env/
    |-- <python虚拟环境>
|-- app/ <项目的模块名称>
    |-- crud/ <前端蓝图>
        |-- __init__.py
        |-- views.py <路由和视图函数文件>
        |-- forms.py <表单类文件, wtforms插件必须项>
        |-- templates <HTML模板>
            |-- static <静态文件夹>
                |-- js <JavaScript 文件夹>
                    |-- 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/<user_id>', 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('<br>' + $('<div/>').text('接收 : ' + msg.data).html());
    });
    socket.on('user_response', function(msg) {
        //console.log(eval(msg.data[0]));
        //$('#users_show').append('<br>' + $('<div/>').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 = "<a href=" +  location.protocol + '//' + document.domain + ':' + location.port + "/crud/websocket-edit/" + value.id + "> 修改</a>";
            delete_url = "<a href=\"javascript:delete_user_" + value.id + "()\">删除</a>";
            var trs = "";
            trs += "<tr><th>" + (n+1) + "</th><td>" + value.username + "</td><td>" + value.email + "</td><td>" + role + "</td><td>" + status + "</td><td>" + edit_url + " | " + delete_url +"</td></tr>";
            tbody += trs;
        })
        $('#users_show').empty();
        $('#users_show').append(tbody);
    });
});

显示结果

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

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

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

总结

  1. 由于 flask 架构具有上下文的限制,在数据库里 增加删改 内容的时候,表格的内容没有变化——尽管局部已经进行了刷新。要想显示变化后的数据库内容,必须得重新启动一下 flask 服务。
  2. 就整体的部署来说,在 flask 项目里添加 websocket 协议,显得项目较重,实现一个局部刷新的功能还是用 ajax 比较简单。
  3. 欢迎大侠能够给我的项目提出修改意见,先行感谢!!!

源码下载

参考

查看原文

赞 9 收藏 12 评论 15

jasonqiao36 赞了文章 · 2017-12-05

如何使用Visual Studio Code开发Django项目

如何获得 Visual Studio Code

访问 http://code.visualstudio.com 下载并安装。

前提条件

安装Python 2.7 及 Python 3.5,Windows下可以安装至 c:\python27 及 c:\python35。注意环境变量PATH的设置,建议只在PATH中添加一个版本的Python路径。Python 安装完后需要使用pip命令安装virtualenv模块,该模块负责隔离不同项目开发之间的模块依赖。

在安装完后,如果是Windows操作系统,可以在%USERPROFILE%下创建pip文件夹,并在该文件夹内创建包含以下内容的pip.ini文件,该文件的作用是让pip命令通过阿里云的镜像站点安装Python模块。

[global]
index-url = http://mirrors.aliyun.com/pypi/simple/

[install]
trusted-host = mirrors.aliyun.com

如何将Visual Studio Code配置成Django项目开发环境

推荐在Windows下使用cmder作为首选命令行工具,配合最新版Git客户端使用,即可在Windows下使用linux命令的移植版本。项目文件夹可为d:\repos。

通常vscode应该以文件夹的形式打开项目,可以在cmder中通过cd命令切换目录至项目文件夹,例如d:\repos\demo,然后再执行 code . 打开项目。

推荐将vscode界面语言配置成英文,方式为在vscode窗口中按下 Ctrl + Shift + P,输入 语言,选择配置语言,再打开的配置文件中,将locale设置为en-US

001.png

安装插件

打开 vscode,按下 Ctrl + Shift + X 打开插件管理侧边栏。依次安装下列插件

Git history
Python
Django Template
jquery Code Snippets
Bootstrap 3 Snippets

创建Django测试项目并进行配置

VS Code提供了两种范围的设置,分别是User及Workspace. 用户级别的设置可以理解为全局设置,其中的设置对任一vscode实例都生效。工作区级别的设置则仅针对当前项目生效,配置文件存放在项目的 .vscode 文件夹中。一般在项目开发中我们使用工作区级别的设置。以下步骤记录了如何建立Django项目的基本目录结构。

  1. 创建项目文件夹,例如 mkdir e:\temp\django_demo,并当前目录切换到该文件夹

  2. 在项目文件夹中使用virtualenv env 创建env文件夹

  3. 创建 requirements.txt 文件,文件中添加以下内容,pylint为python的静态语法检测器,pylint-django 是适用于django项目的语法检查其插件,autopep8 是代码格式化工具

django<1.10
pylint
pylint-django
autopep8

  1. 执行 env\Scripts\activate.bat 激活虚拟环境

  2. 执行 pip install -r requirements.txt 安装相关模块

  3. 执行 django-admin startproject demo . 在当前目录下创建django项目文件夹

  4. 执行 python manage.py startapp web 新建django app

  5. 执行 code . 在当前配置打开项目

  6. 在 vscode 中按下 Ctrl + Shift + P,输入 select,选择 Python: Select Workspace Interpreter,在出现的选项中将Python解析器指向env文件夹中的python

002.png

  1. 在 vscode 中按下 Ctrl + Shift + P,输入 workspace,选择 Preferences: Open Workspace Settings,配置完成后的JSON文件如下

{
    "python.pythonPath": "e:/temp/django_demo/env/scripts/python.exe",
    "python.linting.pylintPath": "pylint",
    "python.linting.pylintArgs": [
        "--load-plugins", "pylint_django"
    ],
    "python.formatting.autopep8Path": "autopep8"
}

成功配置完成后,vscode 会在状态栏中提示代码是否符合 pylint的相关规则。之后我们可以在此基础上继续项目开发。

003.png

查看原文

赞 5 收藏 4 评论 3

jasonqiao36 提出了问题 · 2017-11-16

vim python-mode使用问题

安装了python-mode以后,配置文件都没改,但是一打开.py文件就会报错:

Error detected while processing function pymode#breakpoint#init:
line   17:
E492: Not an editor command: from imp import find_module
line   19:
E110: Missing ')'
E15: Invalid expression: ('wdb', 'pudb', 'ipdb'):
line   20:
E488: Trailing characters:     try:
line   28:
E170: Missing :endfor
Press ENTER or type command to continue

这是什么原因呀,是我的问题还是python-mode本身的问题

关注 2 回答 1

jasonqiao36 提出了问题 · 2017-11-16

解决flask url_for()方法,在js中如何使用

$(function(){
        $('.card_no').click(function(){
            var card_no = $(this).text();
            window.location.href = "{{url_for('card_detail', card_no=card_no) }}";
        });

我想在 url_for 中,传递前面获取的 card_no ,但是这种写法不对,获取不到值。求助大家

关注 4 回答 3

jasonqiao36 提出了问题 · 2017-09-13

解决flask-sqlalchemy查询的问题

一个数据库查询,不太会写,请大家帮忙.

result = db.session.query(A, B).filter(A.device_id==B.device_id).all()

我想要的是, A的device_id的前三位等于B的device_id的前三位。
类似下面的写法(这种写法不对哦):

result = db.session.query(A, B).filter(A.device_id[:3]==B.device_id[:3]).all()

上面的写法不对,但是能表达我的意思了。我应该怎么写呢?

关注 2 回答 1

jasonqiao36 提出了问题 · 2017-08-07

flask-sqlalchemy中distinct的问题

所有手机销售记录中,id是主键。brand是手机的标识,可以重复。

class PhoneSaleRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    brand = db.Column(db.String(20), nullable=False)
    price = db.Column(db.Integer)
    added = db.Column(db.DateTime)

我想查找,每个手机品牌中,最后录入数据库的手机。

现在已经可以通过两步完成这个目标,先找到所有品牌(不重复):

brands = db.session.query(Phone.brand).dinstinct().all()

再根据brand找到对应的phone

for p in brands:
    Phone.query.filter_by(brand=brand).order_by(Phone.added.desc()).first()

但是由于分页的限制,需要将两个SQL合并为一个。
这个该怎么做呀?

关注 1 回答 0

jasonqiao36 回答了问题 · 2017-08-01

flask如何转换并返回list

str(your_list)

关注 2 回答 1

jasonqiao36 提出了问题 · 2017-07-27

解决Python3.6.2使用pip安装包报错

Python3.6.2

创建虚拟环境
python -m venv venv
pip安装包时报错:
pip install flask
错误:

Exception:                                                                                        
Traceback (most recent call last):                                                                
  File "d:\python36-32\lib\site-packages\pip\basecommand.py", line 215, in main                   
    status = self.run(options, args)                                                              
  File "d:\python36-32\lib\site-packages\pip\commands\install.py", line 342, in run               
    prefix=options.prefix_path,                                                                   
  File "d:\python36-32\lib\site-packages\pip\req\req_set.py", line 784, in install                
    **kwargs                                                                                      
  File "d:\python36-32\lib\site-packages\pip\req\req_install.py", line 851, in install            
    self.move_wheel_files(self.source_dir, root=root, prefix=prefix)                              
  File "d:\python36-32\lib\site-packages\pip\req\req_install.py", line 1064, in move_wheel_files  
    isolated=self.isolated,                                                                       
  File "d:\python36-32\lib\site-packages\pip\wheel.py", line 247, in move_wheel_files             
    prefix=prefix,                                                                                
  File "d:\python36-32\lib\site-packages\pip\locations.py", line 141, in distutils_scheme         
    d.parse_config_files()                                                                        
  File "d:\python36-32\lib\distutils\dist.py", line 395, in parse_config_files                    
    parser.read(filename)                                                                         
  File "d:\python36-32\lib\configparser.py", line 697, in read                                    
    self._read(fp, filename)                                                                      
  File "d:\python36-32\lib\configparser.py", line 1015, in _read                                  
    for lineno, line in enumerate(fp, start=1):                                                   
UnicodeDecodeError: 'gbk' codec can't decode byte 0x9d in position 0: incomplete multibyte sequence

关注 3 回答 2

jasonqiao36 回答了问题 · 2017-07-26

解决flask在非视图函数中如何获取配置文件中的参数

from flask import current_app

current_app.config.get('HDFS_IP')

关注 2 回答 1

jasonqiao36 提出了问题 · 2017-07-21

解决如何使用Pycharm调试Flask?

我想在Pycharm中调试Flask。
我现在已经学会在Pycharm中调试普通py模块。
但是遇到flask就懵逼了。
在Flask中,已经设置调试的configuration。调试过程中,老是跳到flask源代码中,完全看不懂源码。
请教大家正确调试Flask项目的姿势!
谢谢 !

关注 4 回答 3

认证与成就

  • 获得 4 次点赞
  • 获得 22 枚徽章 获得 1 枚金徽章, 获得 7 枚银徽章, 获得 14 枚铜徽章

擅长技能
编辑

开源项目 & 著作
编辑

(゚∀゚ )
暂时没有

注册于 2017-02-02
个人主页被 495 人浏览