1 Flask介绍和快速使用
#介绍
Flask是一个基于Python开发并且依赖jinja2模板(DTL)和Werkzeug(wsgiref) WSGI服务的一个微型框架，对于Werkzeug本质是Socket服务端，其用于接收http请求并对请求进行预处理，然后触发Flask框架，开发人员基于Flask框架提供的功能对请求进行相应的处理，并返回给用户，如果要返回给用户复杂的内容时，需要借助jinja2模板来实现对模板的处理，即：将模板和数据进行渲染，将渲染后的字符串返回给用户浏览器
# flask项目的所有代码都可以写在一个py文件中（一般不用），可扩展性高
# 通过集成很多第三方插件实现跟django一样的功能
# wsgiref
from flask import Flask
app = Flask(__name__)
@app.route('/')
def index():
return 'hello flask'
if __name__ == '__main__':
app.run() # 把flask项目跑起来
2 登录，显示用户信息案例
from flask import Flask, request, render_template, redirect, session, jsonify
# flask中的请求对象是全局的request
# render_template 就是django的render
app = Flask(__name__)
# 使用session，一定要设置秘钥，等同于django配置文件中的密码
app.secret_key = 'adfae^^4384045532@@#$#'
app.debug = True # 启用调试模式，修改了代码不用重启，自动重启
USERS = {
1: {'name': '张三', 'age': 18, 'gender': '男', 'text': "道路千万条"},
2: {'name': '李四', 'age': 28, 'gender': '男', 'text': "安全第一条"},
3: {'name': '王五', 'age': 18, 'gender': '女', 'text': "行车不规范"},
}
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'GET':
return render_template('login.html')
else:
# request.form 等同于request.POST
name = request.form.get('user')
password = request.form.get('pwd')
if name == 'lqz' and password == '123':
# 登录成功，重定向到首页
# 把登录信息放到session中
session['name'] = name
return redirect('/')
else:
return jsonify({'code': 101, 'msg': '用户名或密码错误'})
@app.route('/', methods=['GET'])
def index():
if session.get('name'):
return render_template('index.html', user_dict=USERS)
else:
# 没有登录，重定向到登陆页面
return redirect('/login')
# @app.route('/detail/<int:pk>', methods=['GET'])
# def detail(pk):
@app.route('/detail', methods=['GET'],endpoint='detail')
def detail():
print(request.query_string) # b'pk=1&name=lqz&age=19'
pk = int(request.args.get('pk'))
print(pk)
user_detail = USERS.get(pk)
return render_template('detail.html', info=user_detail)
if __name__ == '__main__':
app.run(port=8080)
### 总结
'''
0 当次请求的对象，request对象是全局的，直接导入使用即可
1 前端post提交的数据，从flask中的：request.form中取
2 转换器的使用/detail/<int:pk>
3 模板语法兼容django的dtl，可以直接使用函数加括号，并且可以传参数
4 session也是全局对象，用户登录信息，放到session中
5 新手四件套
'字符串'---》HttpResonse('字符串')
redirect('/')---》redirect('/')
render_template()--->render()
jsonify ---->JsonResponse()
5 前端get请求提交的数据，从request.args
6 request.query_string是get请求 ? 后面的内容，bytes格式
7 @app.route('/detail', methods=['GET'],endpoint='detail') endpoint等同于django路由中的name别名
'''
2.2 detail.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>详细信息 {{info.name}}</h1>
<div>
{{info.text}}
</div>
</body>
</html>
2.3 index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>用户列表</h1>
<table>
{% for k,v in user_dict.items() %}
<tr>
<td>{{k}}</td>
<td>{{v.name}}</td>
<td>{{v['name']}}</td>
<td>{{v.get('name')}}</td>
<td><a href="/detail/{{k}}">查看详细1</a></td>
<td><a href="/detail?pk={{k}}&name=lqz&age=19">查看详细2</a></td>
</tr>
{% endfor %}
</table>
</body>
</html>
2.4 login.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>用户登录</h1>
<form method="post">
<input type="text" name="user">
<input type="text" name="pwd">
<input type="submit" value="登录">{{error}}
</form>
</body>
</html>
3 Flask配置文件
# flask.config.Config
方式一：直接写app.config['DEBUG'] 一般不用
方式二：放到py文件中 一般不用
方式三：通过环境变量 一般不用
app.config.from_envvar("环境变量名称")
app.config.from_json("json文件名称")
JSON文件名称，必须是json格式，因为内部会执行json.loads
app.config.from_mapping({'DEBUG': True})
方式四： 通过类的方式 用的多
app.config.from_object('settings.DevelopmentConfig')
0 函数加装饰器的执行顺序
# flask的路由基于装饰器----》在视图函数上再加装饰器---》加多个装饰器的执行顺序---》登录认证装饰器---》加载router下，先做路由匹配，匹配成功执行被auth包裹的视图函数
1 路由系统
# flask的路由是基于装饰器的---》但是它的本质是app.add_url_rule方法--->类似于django 的paht（url）
1.1 转换器
app.add_url_rule(rule='/index/<string:name>',endpoint='index',view_func=index,methods=['GET'])
# 跟django 2.x后的转换器一样
DEFAULT_CONVERTERS = {
'default': UnicodeConverter,
'string': UnicodeConverter,
'any': AnyConverter,
'path': PathConverter,
'int': IntegerConverter,
'float': FloatConverter,
'uuid': UUIDConverter,
}
1.2 路由系统的本质
"""
# route的源代码
1. decorator = app.route('/',methods=['GET','POST'],endpoint='n1')
def route(self, rule, **options):
def decorator(f):
endpoint = options.pop('endpoint', None)
self.add_url_rule(rule, endpoint, f, **options)
return f
return decorator
2. @decorator
decorator(index)----》本质---》self.add_url_rule(rule, endpoint, f, **options)
3 路由的本质可以写成
app.add_url_rule(rule='/',endpoint='index',view_func=index,methods=['GET'])
4 （了解）endpoint如果不传，是None，也有个默认值---》函数的名字
-如果使用了装饰器，一定要指定endpoint
"""
1.3 app.add_url_rule参数
@app.route和app.add_url_rule参数:
#1 rule, URL规则
#2 view_func, 视图函数内存地址
#3 defaults = None, 默认值, 当URL中无参数，函数需要参数时，使用defaults = {'k': 'v'} 为函数提供参数
#4 endpoint = None, 名称，用于反向生成URL，即： url_for('名称')
#5 methods = None, 允许的请求方式，如：["GET", "POST"]
#6 对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
'''
#7 重定向到指定地址
redirect_to = None,
'''
@app.route('/index/<int:nid>', redirect_to='/home/<nid>')
'''
1.4 支持正则(不用)
#1 写类，继承BaseConverter
#2 注册：app.url_map.converters['regex'] = RegexConverter
# 3 使用：@app.route('/index/<regex("\d+"):nid>') 正则表达式会当作第二个参数传递到类中
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):
"""
路由匹配时，匹配成功后传递给视图函数中参数的值
"""
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()
1.5 CBV源码分析(跟djagno没有区别)
from flask import Flask, url_for
from flask.views import View, MethodView
app = Flask(__name__)
app.debug = True # 热更新---->不停机更新
@app.route(rule='/', methods=['GET'], endpoint='index')
def index():
return 'hello flask'
# 1 写一个类，继承MethodView
class LoginView(MethodView):
def get(self):
return 'get --login'
def post(self):
return 'post---login'
# 2 配置路由,LoginView.as_view()但是必须带一个参数，参数是这个地址的别名
app.add_url_rule('/login', view_func=LoginView.as_view('login'))
# add_url_rule--->等同于django的pathpaht('/',index,name='index')
if __name__ == '__main__':
app.run()
2 模板语法
# 完全兼容DTL，比DTL强大一些，可直接执行函数，还可以传参数
#1.Markup等价django的mark_safe ,
# 2.extends,include一模一样
3 请求与响应
# request对象属性和方法
# request.method 提交的方法
# request.args get请求提及的数据
# request.form post请求提交的数据
# request.values post和get提交的数据总和（一般不用）
# request.cookies 客户端所带的cookie
# request.headers 请求头
# request.path 不带域名，请求路径 http://127.0.0.1:5000/test_if?name=lqz&age=10----》/test_if
# request.full_path 不带域名，带参数的请求路径--->/test_if?name=lqz&age=10
# 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']
# response对象
######新手四件套
# return "字符串"
# return render_template('html模板路径',**{})
# return redirect('/index.html')
# return jsonify({'k1':'v1'})
# 向cookie中写值
# 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
4 session
# 通过全局的session
-设置值：session['key']=value
-取值：session['key']
-删除值：del session['key']
-不用管不同的用户，统一都用全局session取，不会乱
# 一部分源码解读（等同于django中的中间件django.contrib.sessions.middleware.SessionMiddleware）
-这样设置session['key']=value---》在请求走的时候把这个数据，加密，放到cookie中给了前端
-前端带着cookie再一次请求---》请求来的时候，反解出数据，再放到session中，这样你才能使用session['key']
-SecureCookieSessionInterface---》负责干上面那俩事
--save_session:走的时候
-open_session：带着cookie来的时候
5 闪现
# flash 翻译过来的，flash函数
# 作用： 把一些数据放在某个位置，下次要用的时候，直接取出来，取一次就没了，下次请求再取就没了---》可以跨请求
# 放
flash('一坨大便')
# 分类放
flash('一个包子',category='bz')
flash('一个大便',category='db')
flash('又一个大便',category='db')
# 取
res = get_flashed_messages()
#分类取
res = get_flashed_messages(category_filter=['db'])
# 应用场景：在上一次的请求中有些数据，需要带到下次请求中看，就可以使用闪现
# 其实是放在session中等同于 session[ss]='sdsd'
# django也有类似的东西----》message这个app，就是它
6 请求扩展
# 类似于django的中间件，在请求来之前，和走之后执行一些方法
6.1 before_request
类比django中间件中的process_request，在请求收到之前绑定一个函数做一些事情
#基于它做用户登录认证
@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')
6.2 after_request
类比django中间件中的process_response，每一个请求之后绑定一个函数，如果请求没有异常
@app.after_request
def process_response1(response):
print('process_response1 走了')
return response
6.3 before_first_request
第一次请求时,跟浏览器无关
@app.before_first_request
def first():
pass
6.4 teardown_request
每一个请求之后绑定一个函数，即使遇到了异常
@app.teardown_request
def ter(e):
pass
6.5 errorhandler
路径不存在时404，服务器内部错误500
@app.errorhandler(404)
def error_404(arg):
return "404错误了"
6.6 template_global
标签
@app.template_global()
def sb(a1, a2):
return a1 + a2
#{{sb(1,2)}}
6.7 template_filter
过滤器
@app.template_filter()
def db(a1, a2, a3):
return a1 + a2 + a3
#{{ 1|db(2,3)}}
总结：
1 重点掌握before_request和after_request，
2 注意有多个的情况，执行顺序
3 before_request请求拦截后（也就是有return值），response所有都执行
7 蓝图
# blueprint翻译过来的---》蓝图---》划分目录
# 不使用蓝图划分目录，也可以，但是会出现循环导入
# 使用蓝图可以避免
# 使用蓝图：
1 创建蓝图对象blog = Blueprint('blog', __name__)，可以使用自己的静态文件和template
2 以后路由都使用蓝图对象来注册
3 把所有蓝图在app中注册一下，可以指定前缀
app.register_blueprint(blog)
1 fLask-session
# 原生的session把数据加密后放到了cookie中
# 数据库中，redis，文件中。。。。
# pip install flask-session
# 使用了第三方以后，用法跟之前一样，只是在项目启动是，执行一些代码
# 使用的两种方式
# 方式一：
conn = redis.Redis()
app.session_interface = RedisSessionInterface(conn, 'lqz')
# 我自己写的类有save_session和open_session---》操作redis存储和获取
# 方式二：（通用方案：flask集成第三方的通用方案），本质一样，看源码
from flask_session import Session
# 配置文件
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_KEY_PREFIX'] = 'lqz'
app.config['SESSION_REDIS'] = redis.Redis()
Session(app) # 源码在这---》内部就是做了 app.session_interface=RedisSessionInterface(app.config['SESSION_REDIS'])
# RedisSessionInterface源码分析
from flask import Flask, session
from flask_session import RedisSessionInterface
import redis
app = Flask(__name__)
# 方式一：
# conn = redis.Redis()
# app.session_interface = RedisSessionInterface(conn, 'lqz')
# 我自己写的类有save_session和open_session---》操作redis存储和获取
# 方式二：（通用方案：flask集成第三方的通用方案），本质一样，看源码
from flask_session import Session
# 配置文件
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_KEY_PREFIX'] = 'lqz'
app.config['SESSION_REDIS'] = redis.Redis()
Session(app) # 源码在这---》内部就是做了 app.session_interface=RedisSessionInterface(app.config['SESSION_REDIS'])
@app.route('/')
def index():
session['name'] = 'lqz'
return 'hello'
if __name__ == '__main__':
app.run()
2 数据库连接池
# 为什么要有数据库连接池
-随着并发量的越来越大，mysql的连接数也会增大---》我们创建出连接池后，每次从池中获取连接使用，能够避免并发量过大，导致的数据库崩掉的危险
# pip install dbutils
# 创建池对象
from dbutils.pooled_db import PooledDB
# 获得一个数据库连接池对象（）
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='111',
database='cnblogs',
charset='utf8'
)
# 从池中拿出连接
conn = POOL.connection()
cursor = conn.cursor()
cursor.execute('select * from article')
print(cursor.fetchall())
# 了解
-django和flask都是同步框架
-来一个请求，开启线程执行视图函数
-异步框架：sanic，fastapi，只要被async装饰的函数，就是协程函数
@app.route("/")
async def test(request):
await cousor.excute()
return json({"hello": "world"})
# 协程：单线程下实现并发
# 一旦用了异步，所有框架都必须使用异步
-pymysql，redis是同步框架
-不能使用在异步web框架中
-需要使用专门的异步模块
-aiomysql，aioredis，python到目前为止，没有一个比较好的异步orm框架
-django 3.0以后支持异步
-但是django的orm不支持异步
-python的orm框架
-django的orm：同步
-sqlalchemy：同步
-peewee：同步和异步
# aioredis和aiomyql的使用
# 为什么一旦用了异步框架，后面全要用异步
信号(重要)
# 多线程并发安全的问题：多个线程同时操作同一个数据，出现错乱
# 使用同步锁--》修改数据之前先拿到---》开始修改---》释放锁--》别人再拿
# 线程锁----》分布式锁
# 悲观锁和乐观锁
# 同步锁（互斥锁），递归锁（可重入锁），Event事件，Semaphore（信号量）多把锁
############## 内置信号（flask有一些）
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在其中添加数据时，自动触发
# 自定义信号（我们自定义的）
# 内置信号使用
# 内置信号执行使用步骤：
# 模板渲染后信号执行
# 功能：home页面被渲染，记录一条日志
# 第一步：写一个函数
def template_test(*args,**kwargs):
print(args)
print(kwargs)
print('模板渲染完了')
# 第二步：绑定上内置信号：template_rendered
template_rendered.connect(template_test)
# 第三不：触发信号执行（不需要咱们操作）
### 自定义信号的使用###### 自定义信号步骤
# 第0步：自定义一个信号xxxxx = _signals.signal('xxxxx')
# 第一步：写一个函数def xxx_test(*args, **kwargs): print(args) print(kwargs) print('xxx信号触发了')
# 第二步：绑定上自定义的信号信号：xxxxxxxxxx.connect(xxx_test)
# 第三步：触发信号执行（手动触发，在某个位置触发，比如视图函数中）@app.route('/')def index(): xxxxx.send() # 触发信号 return 'hello'
3.3 django 的信号
###### 自定义信号步骤
# 第0步：自定义一个信号
xxxxx = _signals.signal('xxxxx')
# 第一步：写一个函数
def xxx_test(*args, **kwargs):
print(args)
print(kwargs)
print('xxx信号触发了')
# 第二步：绑定上自定义的信号信号：xxxxx
xxxxx.connect(xxx_test)
# 第三步：触发信号执行（手动触发，在某个位置触发，比如视图函数中）
@app.route('/')
def index():
xxxxx.send() # 触发信号
return 'hello'
5 flask-script
# 实现类似于这样的命令：python manage.py runserver
# 自定义命令：
# django 自定义命令
python manage.py initdb xx.xsl article #只要执行这个命令，就向数据库的article表中写入xx.xsl的数据
# pip3 install flask-script
####基本使用
from flask import Flask
app = Flask(__name__)
# 第一步导入
from flask_script import Manager
# 第二步包装
manager = Manager(app)
@app.route('/')
def index():
return 'hello'
if __name__ == '__main__':
# app.run()
# 第三步使用
manager.run()
5.2 自定义命令
######### 自定义命令
@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 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)
6 请求上下文源码分析
请求上下文执行流程（ctx）：
-0 flask项目一启动，有6个全局变量
-_request_ctx_stack：LocalStack对象----》封装了local
-_app_ctx_stack ：LocalStack对象
-request ： LocalProxy对象
-session ： LocalProxy对象
-1 请求来了 app.__call__()---->内部执行：self.wsgi_app(environ, start_response)
-2 wsgi_app()
-2.1 执行：ctx = self.request_context(environ)：返回一个RequestContext对象，并且封装了request(当次请求的request对象)，session
-2.2 执行： ctx.push()：RequestContext对象的push方法
-2.2.1 push方法中中间位置有：_request_ctx_stack.push(self)，self是ctx对象
-2.2.2 去_request_ctx_stack对象的类中找push方法（LocalStack中找push方法）
-2.2.3 push方法源码：
def push(self, obj):
#通过反射找self._local,在init实例化的时候生成的：self._local = Local()
#Local()flask封装的支持线程和协程的local对象
# 一开始取不到stack，返回None
rv = getattr(self._local, "stack", None)
if rv is None:
#走到这，self._local.stack=[],rv=self._local.stack
self._local.stack = rv = []
# 把ctx放到了列表中
#self._local={'线程id1':{'stack':[ctx,]},'线程id2':{'stack':[ctx,]},'线程id3':{'stack':[ctx,]}}
rv.append(obj)
return rv
-3 如果在视图函数中使用request对象，比如：print(request)
-3.1 会调用request对象的__str__方法，request类是：LocalProxy
-3.2 LocalProxy中的__str__方法：lambda x: str(x._get_current_object())
-3.2.1 内部执行self._get_current_object()
-3.2.2 _get_current_object()方法的源码如下：
def _get_current_object(self):
if not hasattr(self.__local, "__release_local__"):
#self.__local() 在init的时候，实例化的，在init中：object.__setattr__(self, "_LocalProxy__local", local)
# 用了隐藏属性
#self.__local 实例化该类的时候传入的local（偏函数的内存地址：partial(_lookup_req_object, "request")）
#加括号返回，就会执行偏函数，也就是执行_lookup_req_object，不需要传参数了
#这个地方的返回值就是request对象(当此请求的request，没有乱)
return self.__local()
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError("no object bound to %s" % self.__name__)
-3.2.3 _lookup_req_object函数源码如下：
def _lookup_req_object(name):
#name是'request'字符串
#top方法是把第二步中放入的ctx取出来，因为都在一个线程内，当前取到的就是当次请求的ctx对象
top = _request_ctx_stack.top
if top is None:
raise RuntimeError(_request_ctx_err_msg)
#通过反射，去ctx中把request对象返回
return getattr(top, name)
-3.2.4 所以：print(request) 实质上是在打印当此请求的request对象的__str__
-4 如果在视图函数中使用request对象，比如：print(request.method):实质上是取到当次请求的reuquest对象的method属性
-5 最终，请求结束执行： ctx.auto_pop(error)，把ctx移除掉
# 总结：ctx对象中放了：当次请求的request对象，app对象，flashes和session
# python的Local对象(java threadlocal)，多个线程可以同时读写这个变量，并且不会有并发安全的问题---》不同线程操作的是自己的数据---》大字典--》key值为线程id号
l=local()
# 线程1
l.name='lqz'
# 线程2
l.name='egon'