【星海出品】flask(三) 组件

Flask是一个基于Python开发并且依赖jinja2模板和Werkzeug WSGI服务的一个微型框架

wsgiref
因为我们不希望接触到TCP连接、HTTP原始请求和响应格式，所以，需要一个统一的接口协议来实现这样的服务器软件，让我们专心用Python编写Web业务。
这个接口就是WSGI：Web Server Gateway Interface，而wsgiref模块就是python基于wsgi协议开发的服务模块。

args 是request的一个属性，其本质是一个Werkzeug依赖包的的immutableMultiDict的对象，用于解析我们传入的查询字符串，immutableMultiDict对象也继承了Dict类，所以可以使用字典的.get()方法来获取，当然了如果我们有获取原生未解析的原生查询字符串的需求，可以使用query_string属性。

@app.route("/get_data",methods=["GET","POST"])
def get_data():# {'username': 'Generalzy', 'password': 'Generalzy'}print(request.json)# ImmutableMultiDict([])print(request.form)# b'{\r\n    "username":"Generalzy",\r\n    "password":"Generalzy"\r\n}'print(request.data)# ImmutableMultiDict([])print(request.args)# CombinedMultiDict([ImmutableMultiDict([]), ImmutableMultiDict([])])print(request.values)# ImmutableMultiDict([])print(request.files)return jsonify({"code":0})

可以在路径内以/<参数名>的形式指定参数，默认接收到的参数类型是string

'''#######################
以下为框架自带的转换器，可以置于参数前将接收的参数转化为对应类型
string 接受任何不包含斜杠的文本
int 接受正整数
float 接受正浮点数
path 接受包含斜杠的文本
########################'''@app.route("/index/<int:id>",)
def index(id):if id == 1:return 'first'elif id == 2:return 'second'elif id == 3:return 'thrid'else:return 'hello world!'if __name__=='__main__':app.run()

from flask import Flask
from flask import request
from flask import render_template
from flask import redirect
from flask import make_responseapp = Flask(__name__)@app.route('/login.html', methods=['GET', "POST"])
def login():# 请求相关信息# request.method  提交的方法# request.args  get请求提及的数据# request.form   post请求提交的数据# request.values  post和get提交的数据总和# request.cookies  客户端所带的cookie# request.headers  请求头# request.json 		json数据# request.data 		二进制原始数据# request.path     不带域名，请求路径# request.full_path  不带域名，带参数的请求路径# request.script_root  # request.url           带域名带参数的请求路径# request.base_url		带域名请求路径# request.url_root      域名# request.host_url		域名# request.host			127.0.0.1:500# request.files# obj = request.files['the_file_name']# obj.save('/var/www/uploads/' + secure_filename(f.filename))# 响应相关信息# return "字符串"# return render_template('html模板路径',**{})# return redirect('/index.html')# return jsonify({'k1':'v1'})# return send_file(文件)# response = make_response(render_template('index.html'))# response是flask.wrappers.Response类型# response.delete_cookie('key')# response.set_cookie('key', 'value')# response.headers['X-Something'] = 'A value'# return responsereturn "内容"if __name__ == '__main__':app.run()

抽象基类,可以用于重写父类

from abc import ABC, abstractmethod

Flask类下url_map属性（一个Map类的实例）

# 自定义转换器类
class RegexConverter(BaseConverter):def __init__(self,url_map,regex):# 重写父类定义方法super(RegexConverter,self).__init__(url_map)self.regex = regexdef to_python(self, value):# 重写父类方法，后续功能已经实现好了print('to_python方法被调用')return value# 将自定义的转换器类添加到flask应用中
# 具体过程是添加到Flask类下url_map属性（一个Map类的实例）包含的转换器字典属性中
app.url_map.converters['re'] = RegexConverter
# 此处re后括号内的匹配语句，被自动传给我们定义的转换器中的regex属性
# value值会与该语句匹配，匹配成功则传达给url映射的视图函数
@app.route("/index/<re('1\d{10}'):value>")
def index(value):print(value)return "Hello World!"if __name__=='__main__':app.run(debug=True)

每个app中都存在一个url_map，这个url_map中包含了url到endpoint的映射；
当request请求传来一个url的时候，会在url_map中先通过rule找到endpoint，然后再在view_functions中根据endpoint再找到对应的视图函数view_func

可以通过view_functions查看到当前endpoint与视图函数的对应情况；
可以通过url_map查看当前url与endpoint的绑定情况；

from flask import Flask, views, url_for
from werkzeug.routing import BaseConverterapp = Flask(import_name=__name__)class RegexConverter(BaseConverter):"""自定义URL匹配正则表达式"""def __init__(self, map, regex):super(RegexConverter, self).__init__(map)self.regex = regexdef to_python(self, value):"""路由匹配时，匹配成功后传递给视图函数中参数的值"""return int(value)def to_url(self, value):"""使用url_for反向生成URL时，传递的参数经过该方法处理，返回的值用于生成URL中的参数"""val = super(RegexConverter, self).to_url(value)return val# 添加到flask中
app.url_map.converters['regex'] = RegexConverter# 使用
@app.route('/index/<regex("\d+"):nid>')
def index(nid):print(url_for('index', nid='888'))return 'Index'if __name__ == '__main__':app.run()

在请求收到之前绑定一个函数做一些事情，
如果有一个写了return返回值，那么其他的before_request不会执行，视图也不会执行。

# 登录认证
@app.before_request
def process_request(*args,**kwargs):if request.path == '/login':return Noneuser = session.get('user_info')if user:return Nonereturn redirect('/login')

每一个请求之后绑定一个函数，请求出现异常不会执行

@app.after_request
def process_response1(response):print('process_response1 走了')return response

第一次请求时,跟浏览器无关

@app.before_first_request
def first():pass

teardown_request

每次请求之后绑定了一个函数，在非debug模式下即使遇到了异常也会执行。

@app.teardown_request 
def ter(error):pass

errorhandler

路径不存在时404，服务器内部错误500，可以自定义监听响应的状态码并处理：

@app.errorhandler(404)
def error_404(arg):return "404错误了"

接口转换

from flask import Flask,redirect,url_forapp = Flask(__name__)@app.route('/index')
def index():# redirect重定位（服务器向外部发起一个请求跳转）到一个url界面；# url_for给指定的函数构造 URL；# return redirect('/hello') 不建议这样做,将界面限死了return redirect(url_for('hello'))@app.route('/hello')
def hello():return 'this is hello fun'if __name__ == '__main__':app.run()

自定义错误

from flask import Flask,render_template,request,abortapp = Flask(__name__)@app.route("/",methods=['GET','POST'])
def index():if request.method == 'GET':return render_template('index.html')elif request.method == 'POST':name = request.form.get('name')password = request.form.get('password')if name == 'zhangsan' and password == '123456':return 'login sucess'else:# abort的用法类似于python中的raise，在网页中主动抛出错误abort(404)return None# 自定义错误处理方法,将404这个error与Python函数绑定
# 当需要抛出404error时，将会访问下面的代码
@app.errorhandler(404)
def handle_404_error(err):# return "发生了错误，错误情况是：%s"%err# 自定义一个界面return render_template('404.html')if __name__ == '__main__':app.run()

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>Title</title>
</head>
<body>
<!-- 注意图片文件需要放在一个静态文件夹static里 -->
<img src="../static/error404.jpg" alt="" width="1428px" height="57px">
</body>
</html>

再查看新闻之前插入登录页

from flask import Flaskapp = Flask(__name__)@app.route('/')
def hello_world():return 'Hello World!'# 定义装饰器函数
def user_login(func):def inner():# 替代登录操作print('登录操作！')# 执行传入的函数对象func()# 此处如果return inner()，那么返回的是inner函数的执行结果# 而使用return inner，则返回的是inner函数return inner# 定义新闻页面视图函数news
def news():print('这是新闻详情页！')
# 将news函数作为参数传给装饰器函数
show_news=user_login(news)
# 因为user_login返回inner函数，所以show_news()==inner()
show_news()
# 打印出show_news的真实函数名（为inner）
print(show_news.__name__)if __name__ == '__main__':app.run(debug=True)

@user_login
# 定义函数news,该函数将自动被传给装饰器做参数
def news():print('这是新闻详情页！')

from flask import Flaskapp = Flask(__name__)@app.route('/')
def hello_world():return 'Hello World!'# 定义装饰器函数
def user_login(func):# inner函数接收参数def inner(*args,**kwargs):print('登录操作！')# 执行传入函数时使用inner接收到的参数func(*args,**kwargs)return inner# 不带参的不受影响
@user_login
def news():print(news.__name__)print('这是新闻详情页！')
news()# 带参的定义时预声明接收的参数
@user_login
def news_list(*args):# 获取元组args的第一个元素page=args[0]print(news_list.__name__)print('这是新闻列表页的第'+str(page)+'页！')
# 传递给args的元组即为(5,)
news_list(5)if __name__ == '__main__':app.run(debug=True)

类视图、装饰器分别通过继承、包装的方式减少了单个flask程序文件里重复代码的出现，实现了程序的优化；
但是这样处理后的文件内，不同功能的代码块（类视图、视图函数）仍然混杂在一起。如果要制作一个非常大型的程序项目，这样不仅会让代码阅读变得十分困难，而且不利于后期维护；
为了解决这一问题，我们需要引入蓝图（flask.Blueprint），用于实现程序功能的模块化；

导入方法：from flask import Blueprint

• 当接收到请求时，Flask会遍历Flask对象下（已注册）的各蓝图对象，比对蓝图对象中记录的url，比对成功则映射到该url绑定的视图函数并返回响应

from flask import Flask
from flask学习 import news,productsapp = Flask(__name__)
@app.route('/')
def hello_world():return 'hello my world !'# 将对应模块下的蓝图对象注册到app中
app.register_blueprint(news.new_list)
app.register_blueprint(products.product_list)if __name__ == '__main__':app.run(debug=True)

如果访问很多接口需要token 验证,可以使用requests.session 状态保持访问

@app.template_global() //全局模板标签
@app.template_filter() //全局模板过滤器

注册蓝图

from view import view_bpapp = Flask(__name__)
app.register_blueprint(view_bp)

蓝图的befort_request，对当前蓝图有效

专门用来存储用户信息的g对象，g的全称的为global

g对象和session的区别
session对象是可以跨request的，只要session还未失效，不同的request的请求会获取到同一个session，但是g对象不是，g对象不需要管过期时间，请求一次就g对象就改变了一次，或者重新赋值了一次.

信号

内置信号

request_started = _signals.signal('request-started')                # 请求到来前执行
request_finished = _signals.signal('request-finished')              # 请求结束后执行before_render_template = _signals.signal('before-render-template')  # 模板渲染前执行
template_rendered = _signals.signal('template-rendered')            # 模板渲染后执行got_request_exception = _signals.signal('got-request-exception')    # 请求执行出现异常时执行request_tearing_down = _signals.signal('request-tearing-down')      # 请求执行完毕后自动执行（无论成功与否）
appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 应用上下文执行完毕后自动执行（无论成功与否）appcontext_pushed = _signals.signal('appcontext-pushed')            # 应用上下文push时执行
appcontext_popped = _signals.signal('appcontext-popped')            # 应用上下文pop时执行
message_flashed = _signals.signal('message-flashed')                # 调用flask在其中添加数据时，自动触发

使用信号

from flask import Flask,signals,render_templateapp = Flask(__name__)# 往信号中注册函数
def func(*args,**kwargs):print('触发信号',args,kwargs)
signals.request_started.connect(func)# 触发信号： signals.request_started.send()
@app.before_first_request
def before_first1(*args,**kwargs):pass@app.before_first_request
def before_first2(*args,**kwargs):pass@app.before_request
def before_first3(*args,**kwargs):pass@app.route('/',methods=['GET',"POST"])
def index():print('视图')return render_template('index.html')if __name__ == '__main__':app.run()

自定义信号

from flask import Flask, current_app, flash, render_template
from flask.signals import _signals
app = Flask(import_name=__name__)# 自定义信号
xxxxx = _signals.signal('xxxxx')def func(sender, *args, **kwargs):print(sender)
# 自定义信号中注册函数
xxxxx.connect(func)
@app.route("/x")
def index():# 触发信号xxxxx.send('123123', k1='v1')return 'Index' if __name__ == '__main__':app.run()

多 APP 的使用

from werkzeug.wsgi import DispatcherMiddleware
from werkzeug.serving import run_simple
from flask import Flask, current_app
app1 = Flask('app01')
app2 = Flask('app02')@app1.route('/index')
def index():return "app01"@app2.route('/index2')
def index2():return "app2"# http://127.0.0.1:5000/index
# http://127.0.0.1:5000/sec/index2
dm = DispatcherMiddleware(app1, {'/sec': app2,
})if __name__ == "__main__":run_simple('localhost', 5000, dm)

flask-script
用于实现类似于django中 python3 manage.py runserver …类似的命令。
安装：pip3 install flask-script

from flask_script import Manager
app = Flask(__name__)
manager=Manager(app)
...
if __name__ == '__main__':manager.run()
#以后在执行，直接：python3 manage.py runserver
#python3 manage.py runserver --help

自定制命令

@manager.command
def custom(arg):"""自定义命令python manage.py custom 123:param arg::return:"""print(arg)@manager.option('-n', '--name', dest='name')
@manager.option('-u', '--url', dest='url')
def cmd(name, url):"""自定义命令（-n也可以写成--name）执行： python manage.py  cmd -n x -u x执行： python manage.py  cmd --x --url x:param name::param url::return:"""print(name, url)

蓝图实际类似于模块分类，bluePrint 目录下是蓝图模块代码文件
需要有一个 ./bluePrint/init.py 文件，这个文件名不能改 ;
还一个具体实现蓝图的方法模块文件如： ./bluePrint/views.py，这个名称随便。

./bluePrint/init.py

from flask import Blueprint
#蓝图命名
blue_task = Blueprint("blue_task", __name__)
#导入蓝图的具体实现模块
from . import views

./bluePrint/views.py

from . import blue_task
from flask import Flask,render_template#这里就使用蓝图命名的路由了 blue_task，就不用app.route进行路由，避免出现循环路由（特别是随着项目模块的增加）
@blue_task.route("/tasks",methods=['POST','GET'])
def index():return "hello world,Blueprint!\n"#这里就使用蓝图命名的路由了 blue_task，就不用app.route进行路由，避免出现循环路由（特别是随着项目模块的增加）
@blue_task.route('/tasks/index',methods=['POST','GET'])
def books():return render_template('task.html')

在bluePrint 目录同级别下也要有一个 ./init.py 文件
这个文件名不能改 和一个启动应用文件如：./bluePrintMain.py，这个名称随便。

./init.py :

from flask import Flask
#创建一个蓝图app
def create_app():app = Flask(__name__)# 导入蓝图模块from .bluePrint import blue_task# 注册蓝图app.register_blueprint(blue_task)return app

./bluePrintMain.py :

from . import create_appapp = create_app()if __name__ == '__main__':#启动appapp.run()

这个时候,访问5000:/tasks/index 就会打开 task.html

./templates/task.html :

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>task</title><style>div {width:200px;height:200px;background: khaki;border:2px solid #000000;}</style>
</head>
<body><div>我是蓝图：tasks</div>
</body>
</html>