flask要点与坑

简介

Flask是一个用Python编写的Web应用程序框架，该框架简单易用、模块化、灵活性高。

该笔记主要记录Flask的关键要点和容易踩坑的地方

Flask 日志配置

Flask 中的自带logger模块（也是python自带的模块），通过简单配置可以实现将日志记录到日志文件中（记录关键日志有助于以后分析问题）；更详细的logging配置可以自行去百度。

# 日志模配置
# coding : utf-8

import os
import logging
from logging.handlers import RotatingFileHandler


def init_app(app):
    '''
    param app : FLask实列（启动时的app）
    '''
	basedir = os.path.abspath(os.path.dirname(__file__))
	# Formatter
	formatter = logging.Formatter('%(asctime)s %(levelname)s %(lineno)s %(message)s')
	# 日志配置
	log_path = LOG_PATH = os.path.join(basedir, 'logs')
	log_info = os.path.join(log_path, 'flask.log')
	if not os.path.exists(log_path):
        os.mkdir(log_path)
	app.config['LOG_PATH'] = log_path
	app.config['LOG_PATH_INFO'] = log_info
	app.config['LOG_FILE_MAX_BYTES'] = 100 * 1024 * 1024

	# 轮转数量是 10 个
	app.config['LOG_FILE_BACKUP_COUNT'] = 10
	# FileHandler Info
	file_handler_info = RotatingFileHandler(filename=log_info)
	file_handler_info.setFormatter(formatter)
	file_handler_info.setLevel(logging.INFO)
	app.logger.addHandler(file_handler_info)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from flask import Flask
from log_hander import init_app

app = Flask(__name__)

init_app(app)

@app.route('/')
def test():
    app.logger.info('hello world')
    return "hello world"
1
2
3
4
5
6
7
8
9
10
11

正常情况下，日志（flask.log）中会输出 ”hello world“信息，但实际情况却没有

这里有个坑，就是在init_app 函数最后需要再加上一段代码：

 app.logger.setLevel(logging.INFO) 
1

这个问题困扰我很长时间

Flask Blueprint

Flask的简单之处就是一个py文件就可以创建一个http服务

from flask import Flask
from log_hander import init_app

app = Flask(__name__)

@app.route('/',methods=["GET"])
def test():
    app.logger.info('hello world')
    return "hello world"
# 添加任意数量的 处理函数
app.run()
1
2
3
4
5
6
7
8
9
10
11

实际项目中不能把所有的业务处理都放到一个py文件中，那么就需要划分模块了。

蓝图（Blueprint）可以帮助我们划分模块（有点类似asp.net mvc、java springboot中的控制器）

一般的做法：建立一个python包，在里面添加我们的模块文件（根据业务划分），每个模块中定义一个蓝图（Blueprint），最后在app中注册蓝图（Blueprint）。

home.py 模块

# coding : utf-8

from flask import Blueprint
from flask import render_template

home_blue=Blueprint('home',__name__)

# 访问路径
# http://ip:port/home
# http://ip:port/home/index
@home_blue.route('/',methods=['GET'])
@home_blue.route('/index',methods=['GET'])
def index():
    '''
    首页
    @return render_template
    '''
    return render_template(
        'index.html',
        title='Home Page'
    )

@home_blue.route('/welcom',methods=['GET'])
def welcomPage():
    return render_template('welcome_iframe.html',
        title='Welcom Page')
    pass
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

login.py 模块

# coding : utf-8
from flask import Blueprint
from flask import render_template

login_blue = Blueprint('login', __name__)

# 访问路径
# http://ip:port/
# http://ip:port/index

@login_blue.route('/', methods=['GET'])
@login_blue.route('/index', methods=['GET'])
def index():
    '''
    登陆首页
    @return render_template
    '''
    if user_id == None:
    	return render_template(
                'login.html',
                title='登录页'
        )


@login_blue.route('/login_user', methods=['POST'])
def user_login():
    '''
    登录
    :return:
    '''
    try:
        # 登录逻辑
        return jsonify(common_tools.get_res_model(None, '验证通过', True))
    except:
        return jsonify(common_tools.get_res_model(None, '验证未通过', False))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

_init_.py

# coding : utf-8
from flaskAdmin.views.home import home_blue
from flaskAdmin.views.login import login_blue

def init_route(app):
    app.register_blueprint(login_blue,url_prefix='/')
    app.register_blueprint(home_blue,url_prefix='/home')
1
2
3
4
5
6
7

app.py中设置

from flask import Flask
from my_moudle import init_route

app = Flask(__name__)
init_route(app)

app.run()
1
2
3
4
5
6
7

templates 过滤器

在Flask中，过滤器（filters）是一种用于处理输入并生成输出的函数，用于在模板渲染过程中转换数据。Flask内置了多种过滤器，同时支持自定义过滤器。这里主要是说过自定义过滤器

app中定义过滤器

使用@app.template_filter(“filter name”) 添加

from flask import Flask
from log_hander import init_app

app = Flask(__name__)

@app.template_filter("format_float") 
def formater_float(val:float):
    '''
    定义一个格式化浮点数据的过滤器
    '''
    return "{:.2f}".format(val)

app.run()
1
2
3
4
5
6
7
8
9
10
11
12
13

使用过滤器

<div>
    {{my_value|format_float}}
div>
1
2
3

Blueprint中定义

使用@blue.app_template_filter(“filter name”)添加

@blue.app_template_filter("format_float")
def formater_float(val:float):
    '''
    定义一个格式化浮点数据的过滤器
    '''
    return "{:.2f}".format(val)
1
2
3
4
5
6

注意：Blueprint中定义的过滤器是全局的，所有模板都可以使用

ORM

sqlalchemy 框架是我的首选。

安装：pip install Flask-SQLAlchemy

model定义

# coding: utf-8
from sqlalchemy import Column, DateTime, String, Text
from sqlalchemy.dialects.mysql import INTEGER
from sqlalchemy.ext.declarative import declarative_base
from flask_sqlalchemy import SQLAlchemy

Base = declarative_base()
metadata = Base.metadata

class User(Base):
    __tablename__ = 'user'
	# 指定ID对应的数据字段（model定义字段与数据字段不一致时）
    ID = Column("uuid",String(64, 'utf8_bin'), primary_key=True) 
    UserCode = Column(INTEGER(11), nullable=False)
    UserRealName = Column(String(64, 'utf8_bin'), nullable=False)
    DelFlag = Column(String(2, 'utf8_bin'), nullable=False)
    PassWord = Column(String(128, 'utf8_bin'), nullable=False)
    Remark = Column(Text(collation='utf8_bin'))
    CreateDate = Column(DateTime, nullable=False, comment='创建日期')
    LastLoginTime = Column(DateTime, comment='最后一次登录日期')
    ValidityOfTime = Column(DateTime, comment='账户有效期')
    UserRole = Column(String(64, 'utf8_bin'), nullable=True)

# ...
db = SQLAlchemy()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

app.py 中配置

from flask import Flask
from db_model import db

app = Flask(__name__)

# 关键步骤
# SQLALCHEMY_DATABASE_URI  数据库连接字符串
app.config["SQLALCHEMY_DATABASE_URI"] = 'mysql+pymysql://usermame:password@host:port/dbname?charset=utf8'
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True # 跟踪对象修改，有点类似EF中的状态跟踪
app.config["SQLALCHEMY_ECHO"] = False # 是否打印SQL日志
db.init(app)

app.run()
1
2
3
4
5
6
7
8
9
10
11
12
13

Session

web离不开session。可以使用Flask-Session模块，也可以自定义实现（原理并不复杂，这里不罗嗦了）

安装：pip install Flask-Session

app.py 中配置

from flask import Flask
from datetime import timedelta

app = Flask(__name__)

app.config["SESSION_TYPE"]='redis' # 需要额外安装redis模块
app.config["SESSION_REDIS"]=Redis(
    host="127.0.0.1",
    port=6379
    )
app.config["SESSION_USE_SIGNER"]=True # 使用字符存储
app.config["SECRET_KEY"]='FLASK_SYSADMIN' # session 加密key
app.config["SESSION_PERMANENT"]=False
app.config["PERMANENT_SESSION_LIFETIME"]=timedelta(seconds=60*20) # session超时时间

app.run()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

使用方式

# 导入session模块
from flask import session

def func():
    # ...
    session['user_id'] = user.ID
    session['user_code'] = user.UserCode
	# ...
1
2
3
4
5
6
7
8

建议将session信息存储的redis中，方便后期分布式部署或作集群时共享登录信息

部署

Flask自带一个服务器，但是也明确指出自带的服务器只能用于开发环境，不能用于生产环境。

linux

Linux环境可以使用uwsgi部署，但是uwsgi需要自己编译。有兴趣的化可以百度查一下，这里不推荐。

windows

windows环境下可以部署到IIS中（IIS最低版本7.0，至少时winserver 2008以后的版本），IIS中部署python web的机会不多，但网上确实有部署教程，有兴趣的可以自行搜索。

uvicorn 、tornado承载

uvicorn 、tornado也是python的一种web框架，是一种异步框架，与Flask结合可以降低部署难度，还可以享受Flask开发带来的便利。

这里演示一下tornado的承载方式：

将Flask主项目放到一个pathon包中，经app的声明迁移到_init_.py 中

runserver.py

# coding: utf-8
from tornado.wsgi import  WSGIContainer
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from flaskAdmin import app,config
import sys
import re

def main():
	HOST = config.APP_HOST
	PORT = config.APP_PORT
	if len(sys.argv) == 3:
		_ip_check=r"^(\d{1,3}\.){3}\d{1,3}$"
		ip_addr = sys.argv[1]
		ip_port = sys.argv[2]
		if re.match(_ip_check,ip_addr) !=None and re.match(r'^[1-9]\d{1,5}$',ip_port)!=None:
			PORT=int(ip_port)
			HOST=ip_addr

	http_server=HTTPServer(WSGIContainer(app))
	http_server.listen(PORT,HOST)
	IOLoop.instance().start()

if __name__ == '__main__':
	main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

使用uvicorn 、tornado承载需要注意一点：如果你的服务需要处理高并发的场景，一定要在服务的前面加一个代理来限制最高并发数量，否则你的服务可能会出现崩溃的情况；这是uvicorn 、tornado一个大坑，两个库都没有做最大并发限制，如果并发超过服务器主机的最大限制，操作系统会报一个 select描述符错，而这两个库并不处理这个异常（针对这个问题stackoverflow论坛上有很多人吐槽）。

https

生成密钥（windows环境下需要自行安装openssl）：

# 生成私钥，按照提示填写内容
openssl genrsa -des3 -out server.key 1024
 
# 生成csr文件 ，按照提示填写内容
openssl req -new -key server.key -out server.csr
 
# Remove Passphrase from key
cp server.key server.key.org 
openssl rsa -in server.key.org -out server.key
 
# 生成crt文件，有效期1年（365天）
openssl x509 -req -days 365 -in server.csr -signkey server.key -out server.crt
1
2
3
4
5
6
7
8
9
10
11
12

Flask app中配置的方式

from flask import Flask
from log_hander import init_app

app = Flask(__name__)

if __name__ == "__main__":
	app.run(host='127.0.0.1',port=10262,
           ssl_context = ("SLL_CRT_PATH","SSL_KEY_PATH")) # 设置ssl 密钥路径
1
2
3
4
5
6
7
8

tornado承载中配置

如果使用tornado承载部署，可以在tornado.httpserver.HTTPServer中设置密钥

http_server=HTTPServer(WSGIContainer(app),
                       ssl_options={
    					   "certfile":"SLL_CRT_PATH",
                           "keyfile":"SSL_KEY_PATH"
						})
1
2
3
4
5