Flask入门三（Flask-session的使用、数据库链接池、wtforms、Flask定制命令、Flask-Cache）

文章目录

一、Flask-session使用
- 1.使用方式一
- 2.使用方式二
- 3.读RedisSessionInterface源码
- 4.flask-session补充
二、数据库连接池
- 1.flask中使用mysql
- 2.上述问题解决
使用数据库连接池
- 1.第三方数据库连接池
- 2.操作数据库不带池版
- 3.池版和非池版压测
三、wtforms
四、Flask定制命令
- 1.使用 flask-script定制命令(老版本，新版本不用了)
- 2.新版本定制命令
- 3.Django中自定制命令
五、Flask-Cache

一、Flask-session使用

Flask内置的Session会把数据加密后保存到浏览器我们自己重写Session类保存到Reids中只需要重写open_session和save_session方法

而在这中间有一个模块就做了这件事，那就是flask-session，以把数据存放到文件、redis、mongodb、关系型数据库等中

安装flask-session

	pip install flask-session

1.使用方式一

	from flask import Flask,session
	app = Flask(__name__)
	app.debug=True
	app.secret_key='jlkdoasfiuz'
	# 只要使用session就需要secret_key
	1.安装flask-session  pip install flask-session

	'使用方式一'
	2.导入(这里我写入到redis缓存数据库中)
	from flask_session import RedisSessionInterface
	3.把app.session_interface替换成RedisSessionInterface的对象
	# 替换了就会走RedisSessionInterface的open_session和save_session
	from redis import Redis
	conn=Redis(host='127.0.0.1',port=6379,db=2)
	# 需要传入的参数 redis, key_prefix, use_signer, permanent, sid_length
	'''
	1.redis 是传入链接的redis库，链接对象
	2.key_prefix    是保存在redis中名称的前缀
	3.use_signer    是如果是False就无需配置secret_key，默认设置True
	4.permanent 是关闭浏览器cookie是否失效
	5.sid_length    是生成session_key的长度，会以cookie形式写入到浏览器cookie中，但是去掉redis中session这个前缀
	去掉前缀就是session_key的长度限制
	'''
	app.session_interface=RedisSessionInterface(
	    redis=conn,key_prefix='session',use_signer=False,
	    permanent=True,sid_length=32
	)
	
	
	@app.route('/set_session')
	def set_session():
	    session['name'] = 'jack'
	    return 'set_session'
	
	
	@app.route('/get_session')
	def get_session():
	    print(session.get('name'))
	    return 'get_session'
	
	
	if __name__ == '__main__':
	    app.run()

2.使用方式二

	from flask import Flask,session
	app = Flask(__name__)
	app.debug=True
	app.secret_key='jlkdoasfiuz'
	# 只要使用session就需要secret_key
	1.安装flask-session  pip install flask-session
	
	'使用方式二'
	2.在flask配置文件中加入配置
	from redis import Redis  # 导入redis
	app.config['SESSION_TYPE'] = 'redis'  # 配置链接的类型
	app.config['SESSION_REDIS']=Redis(host='127.0.0.1',port=6379,db=2)
	# app.config['SESSION_KEY_PREFIX'] = 'session'  # 如果不写，默认以SESSION_COOKIE_NAME作为key
	# app.config.from_pyfile('./settings')  # 第二种导入配置文件方式
	3.导入Session
	from flask_session import Session
	Session(app)  # 核心和方式一一模一样，具体看源码
	
	
	@app.route('/set_session')
	def set_session():
	    session['name'] = 'jack'
	    return 'set_session'
	
	
	@app.route('/get_session')
	def get_session():
	    print(session.get('name'))
	    return 'get_session'
	
	
	if __name__ == '__main__':
	    app.run()

3.读RedisSessionInterface源码

	1.RedisSessionInterface的open_session(在它的父类中)
    def open_session(self, app, request):
    	# -取到前端传入，在cookie中得随机字符串
        sid = request.cookies.get(app.config["SESSION_COOKIE_NAME"])
        if not sid:
            sid = self._generate_sid(self.sid_length)
            # 当sid不为空，把sid传入到session_class得到对象
            return self.session_class(sid=sid, permanent=self.permanent)
        if self.use_signer:  # 用来加密，所以第一种方式的ues_signer最好不要改为False
            try:
                sid = self._unsign(app, sid)
            except BadSignature:
                sid = self._generate_sid(self.sid_length)
                return self.session_class(sid=sid, permanent=self.permanent)
        return self.fetch_session(sid)
	
    def fetch_session(self, sid):
        # 取到随机字符串
        prefixed_session_id = self.key_prefix + sid
        # 从redis中取出key为前缀+随机字符串对应的value值
        value = self.redis.get(prefixed_session_id)
        if value is not None:
            try:
            	# 解密成字符串
                session_data = self.serializer.loads(value)
                # 把解密后的数据，组装到 session对象中
                return self.session_class(session_data, sid=sid)
            except pickle.UnpicklingError:
                return self.session_class(sid=sid, permanent=self.permanent)
        return self.session_class(sid=sid, permanent=self.permanent)
	
	2.RedisSessionInterface的save_session(在它自己内)
   	def save_session(self, app, session, response):
       if not self.should_set_cookie(app, session):
           return
       domain = self.get_cookie_domain(app)
       path = self.get_cookie_path(app)
       
       if not session: # 如果session有值
           if session.modified:  # 如果值被修改过，就把cookie中的删除，并且删除redis中的
               self.redis.delete(self.key_prefix + session.sid)
               response.delete_cookie(
                   app.config["SESSION_COOKIE_NAME"], domain=domain, path=path
               )
           return
       # 
       expiration_datetime = self.get_expiration_time(app, session)
       serialized_session_data = self.serializer.dumps(dict(session))
		# 放到redis中
       self.redis.set(
           name=self.key_prefix + session.sid,
           value=serialized_session_data,
           ex=total_seconds(app.permanent_session_lifetime),  # 过期时间
       )
       # 把session对应的随机字符串放到cookie中
       self.set_cookie_to_response(app, session, response, expiration_datetime)

4.flask-session补充

	- session的前缀如果不传，默认：config.setdefault('SESSION_KEY_PREFIX', 'session:')
    - session过期时间：通过配置，如果不写，会有默认
    	'PERMANENT_SESSION_LIFETIME':           timedelta(days=31),#这个配置文件控制
    -设置cookie时，如何设定关闭浏览器则cookie失效
    	permanent=False
        app.config['SESSION_PERMANENT'] = False

二、数据库连接池

1.flask中使用mysql

	'settings.py'
	SECRET_KEY = 'fdsjakluiz'
	DEBUG = True
	MYSQL_USER = 'root'
	MYSQL_HOST = '127.0.0.1'
	MYSQL_PORT = 3306
	MYSQL_PASSWORD = '1234'
	MYSQL_DATABASE = 'cnblogs'
	JSON_AS_ASCII = False

	'app.py'
	import pymysql.cursors
	from flask import Flask, jsonify
	app = Flask(__name__)
	app.config.from_pyfile('./settings.py')
	
	# pymysql操作mysql
	from pymysql import Connect
	conn = Connect(
	    user=app.config.get('MYSQL_USER'),
	    password=app.config.get('MYSQL_PASSWORD'),
	    host=app.config.get('MYSQL_HOST'),
	    database=app.config.get('MYSQL_DATABASE'),
	    port=app.config.get('MYSQL_PORT'),
	)
	# pymysql.cursors.DictCursor查出来的是列表套字典的形式
	cursor = conn.cursor(pymysql.cursors.DictCursor)
	# cursor = conn.cursor()
	# app.config['JSON_AS_ASCII'] = False  # 前端显示json格式中文
	@app.route('/')
	def articles():
	    cursor.execute('select id,title,author from article limit 10')
	    article_list = cursor.fetchall()  # 拿出所有
	    return jsonify(article_list)
	
	
	if __name__ == '__main__':
	    app.run()

这种方式conn和cursor如果是全局，出现如下问题：

	'上面的  conn和cursor 都是全局的'
	假设极端情况：同时并发两个用户
		-一个用户查询所有文章
	    -一个用户查询所有用户
	    
	'在线程中全局变量是共享的'
	就会出现，第一个线程拿着cursor执行了：
	cursor.execute('select id,title,author from article limit 10')
	然后第二个线程拿着 cursor 执行了：
	cursor.execute('select id,name from user limit 10')
		    
	第一个线程开始执行：（用的全是同一个cursor）
	article_list = cursor.fetchall()
	就会出现查询article的cursor取出来的数据是  用户相关数据---》出现数据错乱

2.上述问题解决

每个人：线程用自己的从conn和cursor，在视图函数中，拿到链接和cursor

	import pymysql.cursors
	from flask import Flask, jsonify
	app = Flask(__name__)
	app.config.from_pyfile('./settings.py')
	
	# pymysql操作mysql
	from pymysql import Connect
	
	@app.route('/')
	def articles():
	    conn = Connect(
	        user=app.config.get('MYSQL_USER'),
	        password=app.config.get('MYSQL_PASSWORD'),
	        host=app.config.get('MYSQL_HOST'),
	        database=app.config.get('MYSQL_DATABASE'),
	        port=app.config.get('MYSQL_PORT'),
	    )
	    cursor = conn.cursor(pymysql.cursors.DictCursor)
	    cursor.execute('select id,title,author from article limit 10')
	    article_list = cursor.fetchall()  # 拿出所有
	    return jsonify(article_list)
	
	
	@app.route('/desc')
	def desc():
	    conn = Connect(
	        user=app.config.get('MYSQL_USER'),
	        password=app.config.get('MYSQL_PASSWORD'),
	        host=app.config.get('MYSQL_HOST'),
	        database=app.config.get('MYSQL_DATABASE'),
	        port=app.config.get('MYSQL_PORT'),
	    )
	    cursor = conn.cursor(pymysql.cursors.DictCursor)
	    cursor.execute('select id,real_desc from article limit 10')
	    article_list = cursor.fetchall()  # 拿出所有
	    return jsonify(article_list)
	
	
	if __name__ == '__main__':
	    app.run()

但是这种如果并发量过高，就会出现连接数过多的问题，mysql的性能就降低了

使用数据库连接池

	'上述操作存在的问题'
	1.原生pymysql操作，最好有一个rom---->sqlalchemy
	2.并发问题：conn和cursor要做成单例，还是每个视图函数一个？
		-如果使用单例，数据会错乱
		-咱们需要，每个视图函数，哪一个链接，如果并发数过多，mysql链接数就会很多，所以使用连接池解决
	
	# django orm操作，一个请求，就会拿到一个mysql的链接，用完后就释放
	'所以想要彻底解决，得使用数据库连接池'
	-限定 mysql链接最多，无论多少线程操作，都是从池中取链接使用

	'解决上面的两个问题'
	-数据库连接池
	-创建一个全局的池
	-每次进入视图函数，从池中取一个连接使用，使用完放回到池中，只要控制池的大小，就能控制mysql连接数

1.第三方数据库连接池

	'pool.py' 在这个文件中配置池也也可以在视图函数中配置
	# 1 安装 pip install dbutils
	# 2 使用：实例化得到一个池对象---》池是单例
	from dbutils.pooled_db import PooledDB
	import 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='1234',
	    database='cnblogs',
	    charset='utf8'
	)
	
	'app.py'  视图函数
	from flask import Flask,jsonify
	app = Flask(__name__)
	app.config.from_pyfile('./settings.py')
	import time
	from pool import POOL
	from pymysql.cursors import DictCursor
	
	## 3 在视图函数中导入使用
	@app.route('/article')
	def article():
	    conn = POOL.connection()
	    cursor = conn.cursor(DictCursor)
	    # 获取10条文章
	    cursor.execute('select id,title,author from article limit 10')
	    time.sleep(1)
	    # 切换
	    res = cursor.fetchall()
	    print(res)
	    return jsonify({'code': 100, 'msg': '成功', 'result': res})
	
	
	if __name__ == '__main__':
	    app.run()

2.操作数据库不带池版

	from flask import Flask,jsonify
	app = Flask(__name__)
	app.config.from_pyfile('./settings.py')
	import time
	## 3 在视图函数中导入使用
	@app.route('/article')
	def article():
	    import pymysql
	    from pymysql.cursors import DictCursor
	    conn = pymysql.connect(user='root',
	                           password="1234",
	                           host='127.0.0.1',
	                           database='cnblogs',
	                           port=3306)
	    cursor = conn.cursor(DictCursor)
	    # 获取10条文章
	    cursor.execute('select id,title,author from article limit 10')
	    time.sleep(1)
	    # 切换
	    res = cursor.fetchall()
	    print(res)
	    return jsonify({'code': 100, 'msg': '成功', 'result': res})
	
	
	if __name__ == '__main__':
	    app.run(port=5001)

3.池版和非池版压测

	压测代码  jmeter工具---》java
	import requests
	from threading import Thread
	
	
	# 没有连接池
	def task():
	    # res = requests.get('http://127.0.0.1:5000/article')  # 带连接池版
	    res = requests.get('http://127.0.0.1:5001/article')  # 不带连接池版
	    print(res.json())
	
	
	if __name__ == '__main__':
	    l = []
	    for i in range(100):
	        t = Thread(target=task)
	        t.start()
	        l.append(t)
	
	    for i in l:
	        i.join()
	
	'''效果是：使用池的连接数明显小，不使用池连接数明显很大'''

查看数据库连接数：show status like '%Threads%';

在这里插入图片描述

三、wtforms

wtforms是一个支持多个web框架的form组件，主要用于对用户请求数据进行验证、渲染错误信息、渲染页面

app.py

from flask import Flask,render_template,request,redirect
from wtforms import Form
from wtforms.fields import simple
from wtforms import validators
from wtforms import widgets

app=Flask(__name__)
app.debug=True

class 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.html

	<!DOCTYPE html>
	<html lang="en">
	<head>
	    <meta charset="UTF-8">
	    <title>Title</title>
	</head>
	<body>
	<h1>登录</h1>
	<form method="post">
	    <p>{{form.name.label}} {{form.name}} {{form.name.errors[0] }}</p>
	    <p>{{form.pwd.label}} {{form.pwd}} {{form.pwd.errors[0] }}</p>
	    <input type="submit" value="提交">
	</form>
	</body>
	</html>

四、Flask定制命令

1.使用 flask-script定制命令(老版本，新版本不用了)

	flask 老版本中，没有命令运行项目，自定制命令
	flask-script 解决了这个问题：flask项目可以通过命令运行，可以定制命令
	
	新版的flask--》官方支持定制命令  click 定制命令，这个模块就弃用了
	flask-migrate 老版本基于flask-script，新版本基于flask-click写的
	
	使用步骤
		-1 pip3 install  Flask-Script==2.0.3
	    -2 pip3 install flask==1.1.4
	    -3 pip3 install markupsafe=1.1.1
		-4 使用
	    from flask_script import Manager
	    manager = Manager(app)
	    if __name__ == '__main__':
	    	manager.run()
	    -5 自定制命令
	    @manager.command
	    def custom(arg):
	        """自定义命令
	        python manage.py custom 123
	        """
	        print(arg)
	        
	    - 6 执行自定制命令
	    python manage.py custom 123

2.新版本定制命令

	from flask import Flask
	import click
	app = Flask(__name__)
	
	自定制命令，通过create-user传入用户名就可以创建一个用户来
	@app.cli.command('create-user')
	@click.argument('name')
	def create_user(name):
	    # from pool import POOL
	    # conn = POOL.connection()
	    # cursor=conn.cursor()
	    # cursor.excute('insert into user (username,password) values (%s,%s)',args=[name,'hello123'])
	    # conn.commit()
	    print(name)
	
	
	命令行中执行
	-flask --app .\Flask定制命令.py:app create-user jack
	-简写成 前提条件式app所在的py文件名叫app.py
	-flask create-user jack
	
	
	@app.route('/')
	def index():
	    return 'index'
	
	
	-运行项目的命令：flask --app .\Flask定制命令.py:app run
	
	if __name__ == '__main__':
	    app.run()

3.Django中自定制命令

	1 app下新建文件夹
		management/commands/
	2 在该文件夹下新建py文件，随便命名（命令名）
	
	3 在py文件中写代码
		from django.core.management.base import BaseCommand
		class Command(BaseCommand):
		    help = '命令提示'
		    def handle(self, *args, **kwargs):
				命令逻辑  
	4 使用命令
		python manage.py  py文件(命令名)

五、Flask-Cache

具体使用可以自寻去官方查看：https://flask-caching.readthedocs.io/en/latest/

	from flask import Flask,render_template
	# 安装 pip install Flask-Caching
	from flask_caching import Cache
	
	config = {
	    "DEBUG": True,
	    "CACHE_TYPE": "SimpleCache",
	    "CACHE_DEFAULT_TIMEOUT": 300
	}
	app = Flask(__name__)
	app.config.from_mapping(config)
	cache = Cache(app)
	
	
	@app.route("/")
	@cache.cached(timeout=50)
	def index():
	    return render_template('index.html')
	
	
	@app.route('/set_cache')
	def set_cache():
	    cache.set('name','xxxx')
	    return 'set_cache'
	
	
	@app.route('/get_cache')
	def get_cache():
	    res = cache.get('name')
	    return res
	
	
	if __name__ == '__main__':
	    app.run()