一、flask介绍
from werkzeug.wrappers import Request, Response@Request.applicationdef hello(request): return Response('Hello World!')if __name__ == '__main__': from werkzeug.serving import run_simple run_simple('localhost', 4000, hello) 三、flask快速使用
from flask import Flask# 实例化产生一个Flask对象app = Flask(__name__)# 将 '/'和视图函数hello_workd的对应关系添加到路由中@app.route('/') # 1. v=app.route('/') 2. v(hello_world)def hello_world(): return 'Hello World!'if __name__ == '__main__': app.run() # 最终调用了run_simple() 登入案例：
from flask import Flask,render_template,request,redirect,session,url_forapp = Flask(__name__)app.debug = Trueapp.secret_key = 'sdfsdfsdfsdf'USERS = { 1:{ 'name':'张三','age':18,'gender':'男','text':"道路千万条"}, 2:{ 'name':'李四','age':28,'gender':'男','text':"安全第一条"}, 3:{ 'name':'王五','age':18,'gender':'女','text':"行车不规范"},}@app.route('/detail/ ',methods=['GET'])def detail(nid): user = session.get('user_info') if not user: return redirect('/login') info = USERS.get(nid) return render_template('detail.html',info=info)@app.route('/index',methods=['GET'])def index(): user = session.get('user_info') if not user: # return redirect('/login') url = url_for('l1') return redirect(url) return render_template('index.html',user_dict=USERS)@app.route('/login',methods=['GET','POST'],endpoint='l1')def login(): if request.method == "GET": return render_template('login.html') else: # request.query_string user = request.form.get('user') pwd = request.form.get('pwd') if user == 'lqz' and pwd == '123': session['user_info'] = user return redirect('http://www.baidu.com') return render_template('login.html',error='用户名或密码错误')if __name__ == '__main__': app.run() #detail.htmlTitle 详细信息 { {info.name}}
{ {info.text}}
#index.htmlTitle 用户列表
| { {k}} | { {v.name}} | { {v['name']}} | { {v.get('name')}} | 查看详细 |
#login.htmlTitle 用户登录
四、配置文件
{ 'DEBUG': get_debug_flag(default=False), 是否开启Debug模式 'TESTING': False, 是否开启测试模式 'PROPAGATE_EXCEPTIONS': None, 'PRESERVE_CONTEXT_ON_EXCEPTION': None, 'SECRET_KEY': None, 'PERMANENT_SESSION_LIFETIME': timedelta(days=31), 'USE_X_SENDFILE': False, 'LOGGER_NAME': None, 'LOGGER_HANDLER_POLICY': 'always', 'SERVER_NAME': None, 'APPLICATION_ROOT': None, 'SESSION_COOKIE_NAME': 'session', 'SESSION_COOKIE_DOMAIN': None, 'SESSION_COOKIE_PATH': None, 'SESSION_COOKIE_HTTPONLY': True, 'SESSION_COOKIE_SECURE': False, 'SESSION_REFRESH_EACH_REQUEST': True, 'MAX_CONTENT_LENGTH': None, 'SEND_FILE_MAX_AGE_DEFAULT': timedelta(hours=12), 'TRAP_BAD_REQUEST_ERRORS': False, 'TRAP_HTTP_EXCEPTIONS': False, 'EXPLAIN_TEMPLATE_LOADING': False, 'PREFERRED_URL_SCHEME': 'http', 'JSON_AS_ASCII': True, 'JSON_SORT_KEYS': True, 'JSONIFY_PRETTYPRINT_REGULAR': True, 'JSONIFY_MIMETYPE': 'application/json', 'TEMPLATES_AUTO_RELOAD': None, }
app.config['DEBUG'] = TruePS： 由于Config对象本质上是字典，所以还可以使用app.config.update(...)
#通过py文件配置app.config.from_pyfile("python文件名称")如：settings.pyDEBUG = Trueapp.config.from_pyfile("settings.py")#通过环境变量配置app.config.from_envvar("环境变量名称")#app.config.from_pyfile(os.environ['YOURAPPLICATION_SETTINGS'])环境变量的值为python文件名称名称，内部调用from_pyfile方法app.config.from_json("json文件名称")JSON文件名称，必须是json格式，因为内部会执行json.loadsapp.config.from_mapping({ 'DEBUG': True})字典格式app.config.from_object("python类或类的路径")app.config.from_object('pro_flask.settings.TestingConfig')settings.py class Config(object): DEBUG = False TESTING = False DATABASE_URI = 'sqlite://:memory:'class ProductionConfig(Config): DATABASE_URI = 'mysql://user@localhost/foo'class DevelopmentConfig(Config): DEBUG = Trueclass TestingConfig(Config): TESTING = True
PS: 从sys.path中已经存在路径开始写
PS: settings.py文件默认路径要放在程序root_path目录，如果instance_relative_config为True，则就是instance_path目录（Flask对象init方法的参数）
五、路由系统
@app.route('/detail/ ',methods=['GET'],endpoint='detail') DEFAULT_CONVERTERS = { 'default': UnicodeConverter, 'string': UnicodeConverter, 'any': AnyConverter, 'path': PathConverter, 'int': IntegerConverter, 'float': FloatConverter, 'uuid': UUIDConverter,} 1. decorator = app.route('/',methods=['GET','POST'],endpoint='n1') def route(self, rule, **options): # app对象 # rule= / # options = {methods=['GET','POST'],endpoint='n1'} def decorator(f): endpoint = options.pop('endpoint', None) self.add_url_rule(rule, endpoint, f, **options) return f return decorator2. @decorator decorator(index) #同理def login(): return '登录'app.add_url_rule('/login', 'n2', login, methods=['GET',"POST"])#与django路由类似#django与flask路由：flask路由基于装饰器，本质是基于：add_url_rule#add_url_rule 源码中，endpoint如果为空，endpoint = _endpoint_from_view_func(view_func)，最终取view_func.__name__（函数名） def auth(func): def inner(*args, **kwargs): print('before') result = func(*args, **kwargs) print('after') return result return innerclass IndexView(views.View): methods = ['GET'] decorators = [auth, ] def dispatch_request(self): print('Index') return 'Index!'app.add_url_rule('/index', view_func=IndexView.as_view(name='index')) # name=endpoint#或者，通常用此方式 class IndexView(views.MethodView): methods = ['GET'] decorators = [auth, ] def get(self): return 'Index.GET' def post(self): return 'Index.POST' app.add_url_rule('/index', view_func=IndexView.as_view(name='index')) # name=endpoint @app.route和app.add_url_rule参数:rule, URL规则view_func, 视图函数名称defaults = None, 默认值, 当URL中无参数，函数需要参数时，使用defaults = { 'k': 'v'}为函数提供参数endpoint = None, 名称，用于反向生成URL，即： url_for('名称')methods = None, 允许的请求方式，如：["GET", "POST"]#对URL最后的 / 符号是否严格要求strict_slashes = None ''' @app.route('/index', strict_slashes=False) #访问http://www.xx.com/index/ 或http://www.xx.com/index均可 @app.route('/index', strict_slashes=True) #仅访问http://www.xx.com/index '''#重定向到指定地址redirect_to = None, ''' @app.route('/index/ ', redirect_to='/home/ ') #子域名访问subdomain = None, ''' #C:\Windows\System32\drivers\etc\hosts 127.0.0.1 www.liuqingzheng.com 127.0.0.1 admin.liuqingzheng.com 127.0.0.1 buy.liuqingzheng.com from flask import Flask, views, url_for app = Flask(import_name=__name__) app.config['SERVER_NAME'] = 'liuqingzheng.com:5000' @app.route("/", subdomain="admin") def static_index(): """Flask supports static subdomains This is available at static.your-domain.tld""" return "static.your-domain.tld" #可以传入任意的字符串，如传入的字符串为aa，显示为 aa.liuqingzheng.com @app.route("/dynamic", subdomain=" ") def username_index(username): """Dynamic subdomains are also supported Try going to user1.your-domain.tld/dynamic""" return username + ".your-domain.tld" if __name__ == '__main__': app.run() 访问： http://www.liuqingzheng.com:5000/dynamic http://admin.liuqingzheng.com:5000/dynamic http://buy.liuqingzheng.com:5000/dynamic ''' #1 写类，继承BaseConverter#2 注册：app.url_map.converters['regex'] = RegexConverter# 3 使用：@app.route('/index/ ') 正则表达式会当作第二个参数传递到类中from flask import Flask, views, url_forfrom werkzeug.routing import BaseConverterapp = Flask(import_name=__name__)class RegexConverter(BaseConverter): """ 自定义URL匹配正则表达式 """ def __init__(self, map, regex): super(RegexConverter, self).__init__(map) self.regex = regex def 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/ ')def index(nid): print(url_for('index', nid='888')) return 'Index'if __name__ == '__main__': app.run() 六、模板
比django中多可以加括号，执行函数，传参数
from flask import Flask,render_template,Markup,jsonify,make_responseapp = Flask(__name__)def func1(arg): return Markup("" %(arg,))@app.route('/')def index(): return render_template('index.html',ff = func1)if __name__ == '__main__': app.run() Title { {ff('六五')}} { {ff('六五')|safe}}
注意：
1.Markup等价django的mark_safe ,
from flask import Flask from flask import request from flask import render_template from flask import redirect from flask import make_response app = Flask(__name__) @app.route('/login.html', methods=['GET', "POST"]) def login(): # 请求相关信息 # request.method # request.args # request.form # request.values # request.cookies # request.headers # request.path # request.full_path # request.script_root # request.url # request.base_url # request.url_root # request.host_url # request.host # 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'}) # 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 response return "内容" if __name__ == '__main__': app.run() 八、session
除请求对象之外，还有一个 session 对象。它允许你在不同请求间存储特定用户的信息。它是在 Cookies 的基础上实现的，并且对 Cookies 进行密钥签名要使用会话，你需要设置一个密钥。 (app.session_interface对象)
设置：session['username'] ＝ 'xxx'删除：session.pop('username', None) 九、闪现
-设置:flash('aaa')
-取值：get_flashed_message()-假设在a页面操作出错，跳转到b页面，在b页面显示a页面的错误信息from flask import Flask,flash,get_flashed_messages,request,redirectapp = Flask(__name__)app.secret_key = 'asdfasdf'@app.route('/index')def index(): # 从某个地方获取设置过的所有值，并清除。 val = request.args.get('v') if val == 'oldboy': return 'Hello World!' flash('超时错误',category="x1") return "ssdsdsdfsd" # return redirect('/error')@app.route('/error')def error(): """ 展示错误信息 :return: """ data = get_flashed_messages(category_filter=['x1']) if data: msg = data[0] else: msg = "..." return "错误信息：%s" %(msg,)if __name__ == '__main__': app.run() 十、请求扩展
#基于它做用户登录认证@app.before_requestdef process_request(*args,**kwargs): if request.path == '/login': return None user = session.get('user_info') if user: return None return redirect('/login')
@app.after_requestdef process_response1(response): print('process_response1 走了') return response 第一次请求时,跟浏览器无关
@app.before_first_requestdef first(): pass
每一个请求之后绑定一个函数，即使遇到了异常
@app.teardown_request def ter(e): pass
@app.errorhandler(404)def error_404(arg): return "404错误了"
#标签@app.template_global()def sb(a1, a2): return a1 + a2#{ {sb(1,2)}} #过滤器@app.template_filter()def db(a1, a2, a3): return a1 + a2 + a3#{ { 1|db(2,3)}} 总结：
2 注意有多个的情况，执行顺序
3 before_request请求拦截后（也就是有return值），response所有都执行
十一、中间件
from flask import Flaskapp = Flask(__name__)@app.route('/')def index(): return 'Hello World!'# 模拟中间件class Md(object): def __init__(self,old_wsgi_app): self.old_wsgi_app = old_wsgi_app def __call__(self, environ, start_response): print('开始之前') ret = self.old_wsgi_app(environ, start_response) print('结束之后') return retif __name__ == '__main__': #把原来的wsgi_app替换为自定义的 app.wsgi_app = Md(app.wsgi_app) app.run() 十二、蓝图
作用：对程序进行目录结构划分
不使用蓝图，自己分文件
目录结构：-templates-views -__init__.py -user.py -order.py-app.py
#app.pyfrom views import appif __name__ == '__main__': app.run()
#init.pyfrom flask import Flask,requestapp = Flask(__name__)#不导入这个不行from . import accountfrom . import orderfrom . import user
#user.pyfrom . import app@app.route('/user')def user(): return 'user' #order.pyfrom . import app@app.route('/order')def order(): return 'order' 目录结构：
-flask_pro -flask_test -__init__.py -static -templates -views -order.py -user.py -manage.py
#__init__.pyfrom flask import Flaskapp=Flask(__name__)from flask_test.views import userfrom flask_test.views import orderapp.register_blueprint(user.us)app.register_blueprint(order.ord)
#manage.pyfrom flask_test import appif __name__ == '__main__': app.run(port=8008)
#user.pyfrom flask import Blueprintus=Blueprint('user',__name__)@us.route('/login')def login(): return 'login' #order.pyfrom flask import Blueprintord=Blueprint('order',__name__)@ord.route('/test')def test(): return 'order test'
1 xxx = Blueprint('account', name,url_prefix='/xxx') ：蓝图URL前缀，表示url的前缀，在该蓝图下所有url都加前缀
2 xxx = Blueprint('account', name,url_prefix='/xxx',template_folder='tpls')：给当前蓝图单独使用templates，向上查找，当前找不到，会找总templates
3 蓝图的befort_request，对当前蓝图有效
4 大型项目，可以模拟出类似于django中app的概念
十二、flash-session
作用：将默认保存的签名cookie中的值 保存到 redis/memcached/file/Mongodb/SQLAlchemy
安装：pip3 install flask-session
使用1：
from flask import Flask,sessionfrom flask_session import RedisSessionInterfaceimport redisapp = Flask(__name__)conn=redis.Redis(host='127.0.0.1',port=6379)#use_signer是否对key签名app.session_interface=RedisSessionInterface(conn,key_prefix='lqz')@app.route('/')def hello_world(): session['name']='lqz' return 'Hello World!'if __name__ == '__main__': app.run() 使用2：
from redis import Redisfrom flask.ext.session import Sessionapp.config['SESSION_TYPE'] = 'redis'app.config['SESSION_REDIS'] = Redis(host='192.168.0.94',port='6379')Session(app)
问题：设置cookie时，如何设定关闭浏览器则cookie失效。
response.set_cookie('k','v',exipre=None)#这样设置即可#在session中设置app.session_interface=RedisSessionInterface(conn,key_prefix='lqz',permanent=False)#一般不用，我们一般都设置超时时间，多长时间后失效 问题：cookie默认超时时间是多少？如何设置超时时间
#源码expires = self.get_expiration_time(app, session)'PERMANENT_SESSION_LIFETIME': timedelta(days=31),#这个配置文件控制
十三、数据库连接池
pymysql连接数据库
import pymysqlconn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123456', db='s8day127db')cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)# cursor.execute("select id,name from users where name=%s and pwd=%s",['maple','123',])cursor.execute("select id,name from users where name=%(user)s and pwd=%(pwd)s",{ 'user':'maple','pwd':'123'})obj = cursor.fetchone()conn.commit()cursor.close()conn.close()print(obj) 数据库连接池版
#setting.pyfrom datetime import timedeltafrom redis import Redisimport pymysqlfrom DBUtils.PooledDB import PooledDB, SharedDBConnectionclass Config(object): DEBUG = True SECRET_KEY = "umsuldfsdflskjdf" PERMANENT_SESSION_LIFETIME = timedelta(minutes=20) SESSION_REFRESH_EACH_REQUEST= True SESSION_TYPE = "redis" PYMYSQL_POOL = PooledDB( creator=pymysql, # 使用链接数据库的模块 maxconnections=6, # 连接池允许的最大连接数，0和None表示不限制连接数 mincached=2, # 初始化时，链接池中至少创建的空闲的链接，0表示不创建 maxcached=5, # 链接池中最多闲置的链接，0和None不限制 maxshared=3, # 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。 blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错 maxusage=None, # 一个链接最多被重复使用的次数，None表示无限制 setsession=[], # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."] ping=0, # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always host='127.0.0.1', port=3306, user='root', password='123456', database='s8day127db', charset='utf8' )class ProductionConfig(Config): SESSION_REDIS = Redis(host='192.168.0.94', port='6379')class DevelopmentConfig(Config): SESSION_REDIS = Redis(host='127.0.0.1', port='6379')class TestingConfig(Config): pass
#utils/sql.pyimport pymysqlfrom settings import Configclass SQLHelper(object): @staticmethod def open(cursor): POOL = Config.PYMYSQL_POOL conn = POOL.connection() cursor = conn.cursor(cursor=cursor) return conn,cursor @staticmethod def close(conn,cursor): conn.commit() cursor.close() conn.close() @classmethod def fetch_one(cls,sql,args,cursor =pymysql.cursors.DictCursor): conn,cursor = cls.open(cursor) cursor.execute(sql, args) obj = cursor.fetchone() cls.close(conn,cursor) return obj @classmethod def fetch_all(cls,sql, args,cursor =pymysql.cursors.DictCursor): conn, cursor = cls.open(cursor) cursor.execute(sql, args) obj = cursor.fetchall() cls.close(conn, cursor) return obj
使用：
obj = SQLHelper.fetch_one("select id,name from users where name=%(user)s and pwd=%(pwd)s", form.data) 十四、wtforms
安装:pip3 install wtforms
from flask import Flask, render_template, request, redirectfrom wtforms import Formfrom wtforms.fields import simplefrom wtforms import validatorsfrom wtforms import widgetsapp = Flask(__name__, template_folder='templates')app.debug = Trueclass LoginForm(Form): # 字段（内部包含正则表达式） name = simple.StringField( label='用户名', validators=[ validators.DataRequired(message='用户名不能为空.'), validators.Length(min=6, max=18, message='用户名长度必须大于%(min)d且小于%(max)d') ], widget=widgets.TextInput(), # 页面上显示的插件 render_kw={ 'class': 'form-control'} ) # 字段（内部包含正则表达式） pwd = simple.PasswordField( label='密码', validators=[ validators.DataRequired(message='密码不能为空.'), validators.Length(min=8, message='用户名长度必须大于%(min)d'), validators.Regexp(regex="^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[$@$!%*?&])[A-Za-z\d$@$!%*?&]{8,}", message='密码至少8个字符，至少1个大写字母，1个小写字母，1个数字和1个特殊字符') ], widget=widgets.PasswordInput(), render_kw={ 'class': 'form-control'} )@app.route('/login', methods=['GET', 'POST'])def login(): if request.method == 'GET': form = LoginForm() return render_template('login.html', form=form) else: form = LoginForm(formdata=request.form) if form.validate(): print('用户提交数据通过格式验证，提交的值为：', form.data) else: print(form.errors) return render_template('login.html', form=form)if __name__ == '__main__': app.run() #login.htmlTitle 登录
from flask import Flask, render_template, request, redirectfrom wtforms import Formfrom wtforms.fields import corefrom wtforms.fields import html5from wtforms.fields import simplefrom wtforms import validatorsfrom wtforms import widgetsapp = Flask(__name__, template_folder='templates')app.debug = Trueclass RegisterForm(Form): name = simple.StringField( label='用户名', validators=[ validators.DataRequired() ], widget=widgets.TextInput(), render_kw={ 'class': 'form-control'}, default='alex' ) pwd = simple.PasswordField( label='密码', validators=[ validators.DataRequired(message='密码不能为空.') ], widget=widgets.PasswordInput(), render_kw={ 'class': 'form-control'} ) pwd_confirm = simple.PasswordField( label='重复密码', validators=[ validators.DataRequired(message='重复密码不能为空.'), validators.EqualTo('pwd', message="两次密码输入不一致") ], widget=widgets.PasswordInput(), render_kw={ 'class': 'form-control'} ) email = html5.EmailField( label='邮箱', validators=[ validators.DataRequired(message='邮箱不能为空.'), validators.Email(message='邮箱格式错误') ], widget=widgets.TextInput(input_type='email'), render_kw={ 'class': 'form-control'} ) gender = core.RadioField( label='性别', choices=( (1, '男'), (2, '女'), ), coerce=int # “1” “2” ) city = core.SelectField( label='城市', choices=( ('bj', '北京'), ('sh', '上海'), ) ) hobby = core.SelectMultipleField( label='爱好', choices=( (1, '篮球'), (2, '足球'), ), coerce=int ) favor = core.SelectMultipleField( label='喜好', choices=( (1, '篮球'), (2, '足球'), ), widget=widgets.ListWidget(prefix_label=False), option_widget=widgets.CheckboxInput(), coerce=int, default=[1, 2] ) def __init__(self, *args, **kwargs): super(RegisterForm, self).__init__(*args, **kwargs) self.favor.choices = ((1, '篮球'), (2, '足球'), (3, '羽毛球')) def validate_pwd_confirm(self, field): """ 自定义pwd_confirm字段规则，例：与pwd字段是否一致 :param field: :return: """ # 最开始初始化时，self.data中已经有所有的值 if field.data != self.data['pwd']: # raise validators.ValidationError("密码不一致") # 继续后续验证 raise validators.StopValidation("密码不一致") # 不再继续后续验证@app.route('/register', methods=['GET', 'POST'])def register(): if request.method == 'GET': form = RegisterForm(data={ 'gender': 2,'hobby':[1,]}) # initial return render_template('register.html', form=form) else: form = RegisterForm(formdata=request.form) if form.validate(): print('用户提交数据通过格式验证，提交的值为：', form.data) else: print(form.errors) return render_template('register.html', form=form)if __name__ == '__main__': app.run() #register.htmlTitle 用户注册
十五、信号
Flask框架中的信号基于blinker，其主要就是让开发者可是在flask请求过程中定制一些用户行为
安装：pip3 install blinker
内置信号：
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_requestdef before_first1(*args,**kwargs): pass@app.before_first_requestdef before_first2(*args,**kwargs): pass@app.before_requestdef before_first3(*args,**kwargs): pass@app.route('/',methods=['GET',"POST"])def index(): print('视图') return render_template('index.html')if __name__ == '__main__': app.wsgi_app app.run() 一个流程中的信号触发点：
a. before_first_requestb. 触发 request_started 信号c. before_requestd. 模板渲染 渲染前的信号 before_render_template.send(app, template=template, context=context) rv = template.render(context) # 模板渲染 渲染后的信号 template_rendered.send(app, template=template, context=context)e. after_requestf. session.save_session()g. 触发 request_finished信号 如果上述过程出错： 触发错误处理信号 got_request_exception.send(self, exception=e) h. 触发信号 request_tearing_down
自定义信号：
from flask import Flask, current_app, flash, render_templatefrom flask.signals import _signalsapp = 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 DispatcherMiddlewarefrom werkzeug.serving import run_simplefrom flask import Flask, current_appapp1 = Flask('app01')app2 = Flask('app02')@app1.route('/index')def index(): return "app01"@app2.route('/index2')def index2(): return "app2"# http://www.maple.com/index# http://www.maple.com/sec/index2dm = 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 Managerapp = Flask(__name__)manager=Manager(app)...if __name__ == '__main__': manager.run()#以后在执行，直接：python3 manage.py runserver#python3 manage.py runserver --help
@manager.commanddef 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 lqz -u http://www.oldboyedu.com 执行： python manage.py cmd --name lqz --url http://www.oldboyedu.com :param name: :param url: :return: """ print(name, url)#有什么用？#把excel的数据导入数据库，定制个命令，去执行