一、Python之flask、Django、Tornado框架
一）django
主要是用来搞快速开发的，他的亮点就是快速开发，节约成本。
正常的并发量不过10000，如果要实现高并发的话，就要对django进行二次开发，比如把整个笨重的框架给拆掉，自己写socket实现http的通信，底层用纯c，c++写提升效率，ORM框架给干掉，自己编写封装与数据库交互的框 架，ORM虽然面向对象来操作数据库，但是它的效率很低，使用外键来联系表与表之间的查询；
二）flask
轻量级，主要是用来写接口的一个框架，实现前后端分离，提升开发效率。第三方组件非常丰富。Flask本身相当于一个内核，其他几乎所有的功能都要用到扩展（邮件扩展Flask-Mail，用户认证Flask-Login），都需要用第三方的扩展来实现。
路由比较特殊：基于装饰器来实现，但是究其本质还是通过add_url_rule来实现。
比如可以用Flask-extension加入ORM、窗体验证工具，文件上传、身份验 证等。Flask没有默认使用的数据库，你可以选择MySQL，也可以NoSQL。
WSGI 工具箱采用 Werkzeug（路由模块），模板引擎则使用 Jinja2。这两个也是Flask框架的核心。
Python最出名的框架要数Django，此外还有Flask、Tornado等框架。
虽然Flask不是最出名的框架，但是Flask应该算是最灵活的框架之一，这也是Flask受到广大开发者喜爱的原因。
三）Tornado
Tornado是一种 Web 服务器软件的开源版本。异步非阻塞框架
Tornado 和现在的主流 Web 服务器框架（包括大多数 Python 的框架）有着明显的区别：它是非阻塞式服务器，而且速度相当快。
得利于其非阻塞的方式和对epoll的运用，Tornado 每秒可以处理数以千计的连接，因此 Tornado 是实时 Web 服务的一个 理想框架。
二、Flask基础学习
一）flask入门
官网：https://flask.palletsprojects.com/en/1.1.x/
中文文档：https://dormousehole.readthedocs.io/en/latest/
1、项目结构
web项目：MVC
model 模型
view 视图
controller 控制器
python：MTV
model 模型
templates 模板----%html
view 视图：起控制作用
2、flask常用三方组件
常用三方组件flask常用组件 - 简书
3、安装flask
pip3 install flaskpip3 install pymysql #连接mysql
pip3 install flask-sqlalchemy # 实现ORM映射
pip3 install flask-migrate #发布命令工具
pip3 install flask_scripts # 管理工具
4、基本使用
from flask import Flask
# 实例化Flask对象
app = Flask(__name__)
# 将'/'和函数index的对应关系添加到路由中
'''
{
'/':index
}
'''
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
#监听用户请求
#若有用户请求到来，则执行app的__call__方法
#app.__call__
app.run()1）参数
- 初始化参数
- import_name
- static_url_path
- static_folder 默认是项目根路径下的static
- template_folder 默认是项目根目录下的templates
2）外部可见的服务器
运行服务器后，会发现只有您自己的电脑可以使用服务，而网络中的其他电脑却 不行。缺省设置就是这样的，因为在调试模式下该应用的用户可以执行您电脑中 的任意 Python 代码。
终端启动
如果您关闭了调试器或信任您网络中的用户，那么可以让服务器被公开访问。 只要在命令行上简单的加上 --host=0.0.0.0 即可:
说明：
- 环境变量FLASK_APP知名flask的启动实例
- flask run -h 0.0.0.0 -p 8000 绑定地址和端口
- flask run --help 获取帮助
- 通过FLASK_ENV 环境变量指明
- export FLASK_ENV=production 运行生产模式，未指明则默认为此方式
- export FLASK_ENV=development 运行开发模式
flask run --host=0.0.0.0 --port=5001在代码里指定
if __name__ == '__main__':
app.run(host="0.0.0.0",port=5000)5、secret_key
1）什么是SECRET_KEY
Flask中有个配置属性叫做SECRET_KEY
2）secret_key 的作用
Flask（以及相关的扩展extension）需要进行加密
所以需要这个密钥SECRET_KEY -> 之所以需要加密，是因为有些涉及到安全的东西，需要加密
3）需要加密的东西
执行代码报错：解决问题的方法——引用secret_key
Internal Server Error
The server encountered an internal error and was unable to complete your request. Either the server is overloaded or there is an error in the application.Flask本身相关的有：session
其它一些第三方的库相关的有：
- Flask-Images（内部可能是图片处理用到的）
- Cookies相关的
- Flask-WTF的CSRF保护
二）配置文件
- 工程配置参数
- app.config.from_object(配置对象)
- 继承 ——> 优点：可以复用
- 缺点：敏感数据暴露
- app.config.from_pyfile('配置文件')
- 优点：独立文件，保护敏感数据
- 缺点：不能继承，文件路径固定，不灵活
- app.config.from_envvar(‘环境变量名’)
- 优点：独立文件，保护敏感数据 文件路径不固定，灵活
- 缺点：不方便，要记得设置环境变量
- 设置环境变量方式
- 系统终端设置：export
- pycharm 设置
- app.config.from_object(配置对象)
1、默认的配置
flask中的配置文件是一个flask.config.Config对象（继承字典）,默认配置为：
{
'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,
}2、增加和变更配置
1、方式一：直接写在代码文件中
app.debug = True
# 或者写成这样
app.config['debug'] = True
app.secret_key = "aaaacccddd"2、方式二 settings.py 一般在项目根路径下
#####写到单独的文件settings.py中
app.config.from_pyfile("python文件名称")
如：
settings.py
DEBUG = True
app.config.from_pyfile("settings.py")
app.config.from_envvar("环境变量名称")
环境变量的值为python文件名称名称，内部调用from_pyfile方法
app.config.from_json("json文件名称")
JSON文件名称，必须是json格式，因为内部会执行json.loads
app.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:'
SQLALCGEMY_DATABASE_URL = 'mysql+pymysql://root@127.0.0.1:3306/flaskstudy
class ProductionConfig(Config):
DATABASE_URI = 'mysql://user@localhost/foo'
class DevelopmentConfig(Config):
DEBUG = True
class TestingConfig(Config):
TESTING = True
PS: 从sys.path中已经存在路径开始写
PS: settings.py文件默认路径要放在程序root_path目录，如果instance_relative_config为True，则就是instance_path目录三）路由系统
1、查看路由信息
- 命令行方法
- flask routes
- 在程序中获取
在应用中url_map属性中保存着整个Flask应用的路由映射信息，可以通过读取这个属性获取路由信息
print(app.url_map)若想在程序中遍历路由信息，可以采用如下方式
for rule in app.url_map.iter_rules():
print('name={} path={}'.format(rule.endpoint, rule.rule))2、请求方式
- GET
- OPTIONS（自带）——>简化版的GET请求，用于询问服务器接口信息的：比如接口允许的请求方式，允许的请求源头域名
- CORS跨域 Django-cors——>中间件中拦截处理了options请求
- HEAD（自带）：简化版的GET请求
- 只返回GET请求处理时的响应头，不返回响应体
- POST
- PUT
- DELETE
- PATCH
3、常见的路由系统
@app.route('/user/<username>')
@app.route('/post/<int:post_id>')
@app.route('/post/<float:post_id>')
@app.route('/post/<path:path>')
@app.route('/login', methods=['GET', 'POST'])所有的路由系统都是基于一个对应关系来处理
DEFAULT_CONVERTERS = {
'default': UnicodeConverter,
'string': UnicodeConverter,
'any': AnyConverter,
'path': PathConverter,
'int': IntegerConverter,
'float': FloatConverter,
'uuid': UUIDConverter,
}动态路由
1）HTML转义
当返回 HTML （ Flask 中的默认响应类型）时，为了防止注入攻击，所有用户提 供的值在输出渲染前必须被转义。使用 Jinja （这个稍后会介绍）渲染的 HTML 模板会自动执行此操作。
在下面展示的 escape() 可以手动转义。因为保持简洁的原 因，在多数示例中它被省略了，但您应该始终留心处理不可信的数据。
from flask import Flask
from markupsafe import escape
app = Flask(__name__)
@app.route("/<name>")
def hello(name):
return f"Hello, {escape(name)}!" 如果一个用户想要提交其名称为 <script>alert("bad")</script> ，那么 宁可转义为文本，也好过在浏览器中执行脚本。
路由中的 <name> 从 URL 中捕获值并将其传递给视图函数
2）变量规则
通过把 URL 的一部分标记为 <variable_name> 就可以在 URL 中添加变量。标记的 部分会作为关键字参数传递给函数。
通过使用 <converter:variable_name> ，可以 选择性的加上一个转换器，为变量指定规则。请看下面的例子:
# -*- coding: utf-8 -*-
# @Time : 2021/7/15 10:12
# @Author : wzs
# @File : flask_v1.py
from flask import Flask
from markupsafe import escape
app = Flask(__name__)
app.debug = True
FLASK_ENV = "development"
from markupsafe import escape
@app.route('/user/<username>')
def show_user_profile(username):
# show the user profile for that user
return f'User {escape(username)}'
@app.route('/post/<int:post_id>')
def show_post(post_id):
# show the post with the given id, the id is an integer
return f'Post {post_id}'
@app.route('/path/<path:subpath>')
def show_subpath(subpath):
# show the subpath after /path/
return f'Subpath {escape(subpath)}'
if __name__ == '__main__':
#监听用户请求
#若有用户请求到来，则执行app的__call__方法
#app.__call__
app.run(host="0.0.0.0",port=5000)
|
（缺省值） 接受任何不包含斜杠的文本 |
|
接受正整数 |
|
接受正浮点数 |
|
类似 |
|
接受 UUID 字符串 |
@app.route('/projects/')
def projects():
return 'The project page'
@app.route('/about')
def about():
return 'The about page' projects 的 URL 是中规中矩的，尾部有一个斜杠，看起来就如同一个文件 夹。访问一个没有斜杠结尾的 URL （ /projects ）时 Flask 会自动进行重 定向，帮您在尾部加上一个斜杠（ /projects/ ）。
about 的 URL 没有尾部斜杠，因此其行为表现与一个文件类似。如果访问这 个 URL 时添加了尾部斜杠（`` /about/ `` ）就会得到一个 404 “未找到” 错 误。这样可以保持 URL 唯一，并有助于搜索引擎重复索引同一页面。
4、注册路由原理
- 装饰器
- url
- method
- endpoint ：url_for
def auth(func):
def inner(*args, **kwargs):
print('before')
result = func(*args, **kwargs)
print('after')
return result
return inner
@app.route('/index.html',methods=['GET','POST'],endpoint='index')
@auth
def index():
return 'Index'
或
def index():
return "Index"
self.add_url_rule(rule='/index.html', endpoint="index", view_func=index, methods=["GET","POST"])
or
app.add_url_rule(rule='/index.html', endpoint="index", view_func=index, methods=["GET","POST"])
app.view_functions['index'] = index
或
def auth(func):
def inner(*args, **kwargs):
print('before')
result = func(*args, **kwargs)
print('after')
return result
return inner
class 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"]
strict_slashes=None, 对URL最后的 / 符号是否严格要求，
如：
@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/<int:nid>', redirect_to='/home/<nid>')
或
def func(adapter, nid):
return "/home/888"
@app.route('/index/<int:nid>', redirect_to=func)
范例：
from flask import Flask
app = Flask(__name__)
app.debug = True
# 公司老新首页跳转
@app.route('/index',methods=['GET','POST'],endpoint='n1',redirect_to="/index2")
def index():
return "公司老首页"
@app.route('/index2',methods=['GET','POST'],endpoint='n2')
def index2():
return "公司新首页"
if __name__ == '__main__':
app.run()子域名
subdomain=None, 子域名访问
from flask import Flask, views, url_for
app = Flask(import_name=__name__)
app.config['SERVER_NAME'] = 'ops.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"
@app.route("/dynamic", subdomain="<username>")
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()5、自定义转换器：正则路由匹配
regex名字固定
from flask import Flask, views, url_for
from werkzeug.routing import BaseConverter
app = 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):
"""
路由匹配时，正则匹配成功后传递给视图函数中参数的值
:param value:
:return:
"""
return int(value)
def to_url(self, value):
"""
使用url_for反向生成URL时，传递的参数经过该方法处理，返回的值用于生成URL中的参数
:param value:
:return:
"""
val = super(RegexConverter, self).to_url(value)
return val
# 第二步：添加到flask中
app.url_map.converters['regex'] = RegexConverter
# 第三步：zai
@app.route('/index/<regex("\d+"):nid>')
def index(nid):
print(url_for('index', nid='888'))
return 'Index'
if __name__ == '__main__':
app.run()Mobile转换器
from flask import Flask
from werkzeug.routing import BaseConverter
app = Flask(__name__)
app.debug = True
# /user/123
# @app.route('/users/<user_id>')
# 等价于
# @app.route('/users/<string:user_id>')
@app.route('/users/<int:user_id>')
def get_users_data(user_id):
print(type(user_id))
return 'get users {}'.format(user_id)
# 定义转换器
class MobileConverter(BaseConverter):
regex = r'1[3-9]\d{9}'
# 将自定义转换器告知Flask应用
app.url_map.converters['mobile'] = MobileConverter
# 在使用转换器的地方定义使用
@app.route('/sms_code/<mobile:mob_num>')
def send_sms_code(mob_num):
print(type(mob_num))
# return 'send sms code to {}'.format(mob_num)
return f'send sms code to {mob_num}'
if __name__ == '__main__':
app.run()四）模板template
1、模板的使用
Flask使用的是Jinja2模板，所以其语法与Django无差别
2、自定义模板方法
Flask中自定义模板方法的方式和Bottle相似，创建一个函数并通过函数形式传入render_template
防XSS攻击，若想标记它时，前使用 |safe，后端使用Markup
注意：Markup等价django的mark_safe
1)后端代码
from flask import Flask,render_template,Markup
app = Flask(__name__)
app.debug = True
def func1(arg):
return Markup("<input type='text' value='%s' />" %(arg,))
@app.route('/')
def index():
return render_template('s7.html', ff = func1)
# return render_template('s7.html', **data) # 所有数据
if __name__ == '__main__':
app.run()2）前端代码
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
{{ ff('侠客')|safe }}
</body>
</html>3）宏定义
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
{% macro input(name, type='text', value='') %}
<input type="{{ type }}" name="{{ name }}" value="{{ value }}">
{% endmacro %}
{{ input('n1') }}
{% include 'tp.html' %}
<h1>asdf{{ v.k1}}</h1>
</body>
</html>五）请求和响应
1、需求
在视图编写中需要读取客户端请求携带的数据时，如何才能正确的取出数据呢？
请求携带的数据可能出现在HTTP报文中的不同位置，需要使用不同的方法来获取参数。
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 "字符串"
# 返回模板 render_template
# return render_template('html模板路径',**{})
# 重定向redirect
# return redirect('/index.html')
# body 返回JSON 的简便方法 ：jsonify一、原来的操作
import json
return json.dumps({}) # 返回的还是文本
二、flask的jsonify
return josonify
# 1 转换成json格式化字符串
# 2 设置了响应头Content-Type: application/json
# 可以返回一个元组，这样的元组必须是(response, status, headers)的形式，且至少包含一个元素。status值会覆盖状态代码，headers可以是一个列表或字典，作为额外的消息标头值
# return '状态码为 666', 666, {'Itcast':'Python'}
# 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()六）cookie和session
1、cookie
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# __author__ = "wzs"
#2021/7/16
from flask import Flask, make_response, request
app = Flask(__name__)
@app.route('/cookie')
def set_cookie():
resp = make_response('set cookie ok')
# 设置有效期
resp.set_cookie('username', 'wzs', max_age = 3600)
return resp
@app.route('/get_cookie')
def ger_cookie():
res = request.cookies.get('username')
return res
@app.route('/delete_cookie')
def delete_cookie():
respose = make_response('delete cookie ok')
respose.delete_cookie('username')
return respose
if __name__ == '__main__':
app.run(port=5001, debug=True)2、session的使用
除请求对象之外，还有一个 session 对象。它允许你在不同请求间存储特定用户的信息。它是在 Cookies 的基础上实现的，并且对 Cookies 进行密钥签名要使用会话，你需要设置一个密钥。
设置：session['username'] ＝ 'xxx'
删除：session.pop('username', None)1）基本使用
from flask import Flask, session, redirect, url_for, escape, request
app = Flask(__name__)
@app.route('/')
def index():
if 'username' in session:
return 'Logged in as %s' % escape(session['username'])
return 'You are not logged in'
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
session['username'] = request.form['username']
return redirect(url_for('index'))
return '''
<form action="" method="post">
<p><input type=text name=username>
<p><input type=submit value=Login>
</form>
'''
@app.route('/logout')
def logout():
# remove the username from the session if it's there
session.pop('username', None)
return redirect(url_for('index'))
# set the secret key. keep this really secret:
app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'2）自定义session
pip3 install Flask-Session
run.py
from flask import Flask
from flask import session
from pro_flask.utils.session import MySessionInterface
app = Flask(__name__)
app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
app.session_interface = MySessionInterface()
@app.route('/login.html', methods=['GET', "POST"])
def login():
print(session)
session['user1'] = 'alex'
session['user2'] = 'alex'
del session['user2']
return "内容"
if __name__ == '__main__':
app.run()
session.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import uuid
import json
from flask.sessions import SessionInterface
from flask.sessions import SessionMixin
from itsdangerous import Signer, BadSignature, want_bytes
class MySession(dict, SessionMixin):
def __init__(self, initial=None, sid=None):
self.sid = sid
self.initial = initial
super(MySession, self).__init__(initial or ())
def __setitem__(self, key, value):
super(MySession, self).__setitem__(key, value)
def __getitem__(self, item):
return super(MySession, self).__getitem__(item)
def __delitem__(self, key):
super(MySession, self).__delitem__(key)
class MySessionInterface(SessionInterface):
session_class = MySession
container = {}
def __init__(self):
import redis
self.redis = redis.Redis()
def _generate_sid(self):
return str(uuid.uuid4())
def _get_signer(self, app):
if not app.secret_key:
return None
return Signer(app.secret_key, salt='flask-session',
key_derivation='hmac')
def open_session(self, app, request):
"""
程序刚启动时执行，需要返回一个session对象
"""
sid = request.cookies.get(app.session_cookie_name)
if not sid:
sid = self._generate_sid()
return self.session_class(sid=sid)
signer = self._get_signer(app)
try:
sid_as_bytes = signer.unsign(sid)
sid = sid_as_bytes.decode()
except BadSignature:
sid = self._generate_sid()
return self.session_class(sid=sid)
# session保存在redis中
# val = self.redis.get(sid)
# session保存在内存中
val = self.container.get(sid)
if val is not None:
try:
data = json.loads(val)
return self.session_class(data, sid=sid)
except:
return self.session_class(sid=sid)
return self.session_class(sid=sid)
def save_session(self, app, session, response):
"""
程序结束前执行，可以保存session中所有的值
如：
保存到resit
写入到用户cookie
"""
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
httponly = self.get_cookie_httponly(app)
secure = self.get_cookie_secure(app)
expires = self.get_expiration_time(app, session)
val = json.dumps(dict(session))
# session保存在redis中
# self.redis.setex(name=session.sid, value=val, time=app.permanent_session_lifetime)
# session保存在内存中
self.container.setdefault(session.sid, val)
session_id = self._get_signer(app).sign(want_bytes(session.sid))
response.set_cookie(app.session_cookie_name, session_id,
expires=expires, httponly=httponly,
domain=domain, path=path, secure=secure)
3）第三方session
#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
pip3 install redis
pip3 install flask-session
"""
from flask import Flask, session, redirect
from flask.ext.session import Session
app = Flask(__name__)
app.debug = True
app.secret_key = 'asdfasdfasd'
app.config['SESSION_TYPE'] = 'redis'
from redis import Redis
app.config['SESSION_REDIS'] = Redis(host='192.168.0.94',port='6379')
Session(app)
@app.route('/login')
def login():
session['username'] = 'alex'
return redirect('/index')
@app.route('/index')
def index():
name = session['username']
return name
if __name__ == '__main__':
app.run()
七）message闪现
message是一个基于Session实现的用于保存数据的集合，其特点是：使用一次就删除。
应用：对临时数据操作，如：显示错误信息
1、基本实现
from flask import Flask,flash,get_flashed_messages
app = Flask(__name__)
# app.secret_key = "ahhjhh227"
@app.route('/get')
def get():
# 从某个地方获取设置过的所有值，并清除
data = get_flashed_messages()
print(data)
return 'get'
@app.route('/set')
def set():
# 向某个地方设置一个值
flash("维纳斯")
return 'set'
if __name__ == '__main__':
app.run()2、实现url错误跳转
session基于用户隔离的，不会发生数据错乱
基于session实现的
from flask import Flask,flash,get_flashed_messages,request,redirect
app = Flask(__name__)
app.secret_key = "ahhjhh227"
# 需求，当遇到错误时，返回错误信息
@app.route('/index')
def index():
# 从某个地方获取设置过的所有值，并清除
val = request.args.get('v')
if val == 'ops':
return 'index'
flash('超时错误',category="x1") #取指定信息
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()
八）蓝图blueprint
1、需求
在一个Flask应用项目中，若业务视图过多，可否将以某种方式划分出的业务单元单独维护，将每个单元用到的视图、静态文件、模板文件等独立分开？
例如：从业务角度上，可将整个应用划分为用户模块单元、商品模块单元、订单模块单元，如何分别开发这些不同单元，并最终整合到一个项目应用中？
2、蓝图简简述
在Flask中，使用蓝图blueprint来分模块组织管理
蓝图实例可以理解为是一个存储一组视图方法的容器对象，其具有如下特点
- 一个应用可以具有多个blueprint
- 可以将一个blueprint注册到任何一个未使用的URL下，比如 /usr，/goods
- blueprint可以单独具有自己模板、静态文件或其他通用操作方法，它并不适合必须要应用的视图和函数的
- 在一个应用初始化时，就应该要注册需要使用的blueprint
但是一个blueprint并不是一个完整的应用，它不能独立于应用运行，而必须要注册到某一应用中
3、使用方式
1）使用蓝图的三个步骤
- 创建一个蓝图对象
user_bp = Blueprint('user', __name__)- 在这个蓝图对象上进行操作，注册路由，指定静态文件夹，注册模板过滤器
@user_bp.route('/profile')
def get_profile():
return 'user profile'- 在应用对象上注册这个蓝图对象
app.register_blueprint(user_bp) 2）单文件蓝图对象
可以将创建蓝图对象与定义视图放在一个文件中
3）目录（包）蓝图对象
对于一个打算包含一个文件的蓝图，通常将创建蓝图对象放到Python包的__init__.py文件中
4、目标：构造程序目录
蓝图的包放在__init__.py文件中
- 批量url
- 模板路径、静态文件路径
- 请求扩展
- 针对APP
- 针对某个蓝图
5、蓝图用于为应用提供目录划分：
小型应用程序：示例
大型应用程序：示例
1）扩展用法：
-
- 蓝图URL前缀：xxx = Blueprint('account', __name__,url_prefix='/xxx')
- 蓝图子域名：xxx = Blueprint('account', __name__,subdomain='admin')
# 前提需要给配置SERVER_NAME： app.config['SERVER_NAME'] = 'ops.com:5000'
# 访问时：admin.ops.com:5000/login.html - 蓝图内部静态文件目录
和应用对应不同，蓝图对象创建时不会默认注册静态目录的路由。需要我们在创建时，制定static_folder参数
admin = Blueprint("admin", __name__, static_folder='static_admin')- 蓝图内部的模板文件目录
admin = Blueprint("admin", __name__, template_folder='template_admin'
6、应用设计的选择
- 一个应用程序（蓝图）：小型应用程序
- N个应用程序（蓝图）：大型应用程序
蓝图名称要和函数名称重复
7、问题说明
- 导入视图放到最后的原因
- 循环引用
views.py
from . import goods_bp
@goods_bp.route('/goods')
def get_goods():
return 'get goods'__init__.py
from flask import Blueprint
# 创建蓝图对象
goods_bp = Blueprint('goods', __name__)
from . import views #此处- Flask Debug模式的作用
- 支持静态文件
- 后端出现错误，会直接返回真实的错误信息给前端
- 修改代码后，自动重启开发服务器
- 支持options请求不等价于实现CORS跨域解决方案
九）请求钩子(中间层/件)
请求的处理过程：pre_process -> view -> after_process
request请求支持处理流程：m1.pre_process() -> m2.pre_process() -> m3.pre_process() -> view -> m3.after_process -> m2.after_process -> m1.after_process
中间件处理 不区分具体是哪个视图，对所有视图通通生效
在客户端和服务器交互的过程中，有些准备工作或扫尾工作需要处理，比如
- 在请求开始时，建立数据库连接
- 在请求开始时，根据需求进行权限校验
- 在请求结束时，指定数据的交互格式
为了让每个视图函数避免编写重复功能的代码，Flask通过了通用设施的功能，即请求钩子
请求钩子是通过装饰器的形式实现，Flask支持如下四种请求钩子：
- before_first_request
- 在处理第一个请求前执行
- before_request
- 在每次请求前执行
- 若在某装饰的函数中返回了一个响应，视图函数将不再被调用
- after_request
- 若没有抛出错误，在每次请求后执行
- 接受一个参数：视图函数做出的响应
- 在此函数中可以对响应值在返回之前做最后一步修改处理
- 需要将参数中的响应在此参数中进行返回
- teardown_request
- 在每次请求后执行
- 接受一个参数：错误信息，若有相关错误抛出
1、基于before_request做用户登录认证
from flask import Flask,request,session,redirect
app = Flask(__name__)
@app.before_request
def process_request(*args, **kwargs):
if request.path == '/login':
return None
user = session.get('user_info')
if user:
return None
return redirect('/login')2、执行顺序
from flask import Flask
app = Flask(__name__)
app.debug = True
app.secret_key = 'suhdjakjdsh'
@app.before_request
def process_request1(*args, **kwargs):
print("process_request1进来了")
return "拦截"
@app.before_request
def process_request2(*args, **kwargs):
print("process_request2进来了")
@app.after_request
def process_response1(response):
print("process_response1走了")
return response
@app.after_request
def process_response2(response):
print("process_response2走了")
return response
# 自定义错误
@app.errorhandler(404)
def error(arg):
return "404错误了"
@app.route('/index',methods=['GET'])
def index():
print("index函数")
return "index"
if __name__ == '__main__':
app.run()3、请求拦截后，response都执行
from flask import Flask
app = Flask(__name__)
app.debug = True
app.secret_key = 'suhdjakjdsh'
@app.before_request
def process_request1(*args, **kwargs):
print("process_request1进来了")
return "拦截"
@app.before_request
def process_request2(*args, **kwargs):
print("process_request2进来了")
@app.after_request
def process_response1(response):
print("process_response1走了")
return response
@app.after_request
def process_response2(response):
print("process_response2走了")
return response
@app.route('/index',methods=['GET'])
def index():
print("index函数")
return "index"
if __name__ == '__main__':
app.run()4、异常处理：自定义错误
1）HTTP异常主动抛出
- abort 方法
- 抛出一个给定状态代码的HTTPException或者指定响应，例如想要一个页面未找到异常来终止请求，你可以调用abort(404)
- 参数
- code - HTTP的错误状态码（只能抛出HTTP协议的错误状态码）
2）捕获错误
- errorhandler 装饰器
- 注册一个错误处理程序，当程序抛出指定错误状态码时，就会调用该装饰器所装饰的方法
- 参数
- code_or_exception - HTTP 的错误状态码或指定异常
@app.errorhandler(500)
def internal_server_error(e):
return '服务器搬家了！'
# 捕获指定异常
@app.errorhandler(ZeroDivisionError)
def zero_division_error(e):
return '除数不能为0'
# 自定义错误
@app.errorhandler(404)
def error(arg):
return "404错误了"5、模板函数 调用方式：{{sb(1,2)}} {{ 1|db(2,3)}}
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import Flask, Request, render_template
app = Flask(__name__, template_folder='templates')
app.debug = True
@app.before_first_request
def before_first_request1():
print('before_first_request1')
@app.before_first_request
def before_first_request2():
print('before_first_request2')
@app.before_request
def before_request1():
Request.nnn = 123
print('before_request1')
@app.before_request
def before_request2():
print('before_request2')
@app.after_request
def after_request1(response):
print('before_request1', response)
return response
@app.after_request
def after_request2(response):
print('before_request2', response)
return response
@app.errorhandler(404)
def page_not_found(error):
return 'This page does not exist', 404
@app.template_global()
def sb(a1, a2):
return a1 + a2
@app.template_filter()
def db(a1, a2, a3):
return a1 + a2 + a3
@app.route('/')
def hello_world():
return render_template('hello.html')
if __name__ == '__main__':
app.run()十）上下文管理
上下文：即语境，语意，在程序中可以理解为在代码执行到某一时刻时，根据之前代码所做的操作以及山该文即将要执行的逻辑，可以决定在当前时刻下可以使用到的变量，或者可以完成的事情。
Flask中有两种上下文，请求上下文和应用上下文。
Flask中上下文对象：相当于一个容器，保存了Flask程序运行过程中的一些信息。
1、请求上下文（request context）
思考：在视图函数中，如何取到当前请求的相关数据？比如：请求地址，请求方式，cookie等等
在flask中，可以直接在视图函数中使用request这个对象进行获取相关数据，而request就是请求上下文的对象，保存了当前本次请求的相关数据，请求上下文对象有：request、session
- request
- 封装了HTTP请求的内容，针对的是http请求。举例：user=request.args.get('user')，获取的是get请求参数
- session
- 用来记录请求会话的信息，针对的是用户信息。举例：session['name']=user.id，可以记录用户信息。还可以通过session.get('name')获取用户信息
2、应用上下文（application context）
字面上是应用上下文，但它不是一直存在的，它只是request context中的一个对app的代理，所谓的local proxy。
应用上下文的作用主要是帮助request获取当前的应用。
应用上下文对象有：current_app，g
1）current_app
应用程序上下文，用于存储应用程序中的变量，可以通过current_app.name打印当前app的名称，也可以在current_app中存储一些变量，例如
- 应用的启动脚本是哪个文件，启动时指定了哪些参数
- 加载了哪些配置文件，导入了哪些配置
- 连了哪些数据库
- 在哪些public的工具类、常量
- 应用跑在哪个机器上，IP多少，内存多大
作用：current_app 就是当前运行的flask app，在代码不方便直接操作flask的对象时，可以操作current_app就等价于操作flask app对象
2）g对象
g作为flask程序全局的一个临时变量，充当中间媒介的作用，我们可以通过它在一个请求调用的多个函数间传递一些数据。每次请求都会重设这个变量
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# __author__ = "wzs"
#2021/7/17
from flask import Flask, g
app = Flask(__name__)
def db_query():
user_id = g.user_id
user_name = g.user_name
print(f'user_id={user_id} user_name={user_name}')
@app.route('/')
def get_user_profile():
g.user_id = 123
g.user_name = 'wzs'
db_query()
return 'you are great!'
if __name__ == '__main__':
app.run()3、上下文实现的原理
Threadlocal 线程局部变量
from flask import request
request ->全局变量
threading.local 对象，用于为每个线程开辟一块空间来保存它独有的值
1）现状：三种情况
- 单进程单线程，基于全局变量做
- 单进程多线程，threading.local对象
- 单进程单线程（多协程），threading.local对象做不到
2）对此现状的抉择
- 以后不支持协程：threading.local对象
- 支持：自定义类似threading.local对象（支持协程）
3）自定义类似threading.local对象
greenlet import getcurrent as get_ident
from greenlet import getcurrent as get_ident
def release_local(local):
local.__release_local__()
class Local(object):
__slots__ = ('__storage__', '__ident_func__')
def __init__(self):
# self.__storage__ = {}
# self.__ident_func__ = get_ident
object.__setattr__(self, '__storage__', {})
object.__setattr__(self, '__ident_func__', get_ident)
def __release_local__(self):
self.__storage__.pop(self.__ident_func__(), None)
def __getattr__(self, name):
try:
return self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
def __setattr__(self, name, value):
ident = self.__ident_func__()
storage = self.__storage__
try:
storage[ident][name] = value
except KeyError:
storage[ident] = {name: value}
def __delattr__(self, name):
try:
del self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
class LocalStack(object):
def __init__(self):
self._local = Local()
def __release_local__(self):
self._local.__release_local__()
def push(self, obj):
"""Pushes a new item to the stack"""
rv = getattr(self._local, 'stack', None)
if rv is None:
self._local.stack = rv = []
rv.append(obj)
return rv
def pop(self):
"""Removes the topmost item from the stack, will return the
old value or `None` if the stack was already empty.
"""
stack = getattr(self._local, 'stack', None)
if stack is None:
return None
elif len(stack) == 1:
release_local(self._local)
return stack[-1]
else:
return stack.pop()
@property
def top(self):
"""The topmost item on the stack. If the stack is empty,
`None` is returned.
"""
try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
stc = LocalStack()
stc.push(123)
v = stc.pop()
print(v)
object.__setattr__(self, 'storage', {})
self.storage = {}对象.xx
def __setattr__(self, key, value):
print(key,value)