系列文章目录

第三章数据库连接池、wtfroms、信号、多app应用、flask-script、flask请求上下文分析

文章目录

系列文章目录
一、数据库连接池
- 安装 dbutils
- dbutils使用
二、wtfroms
- 安装
- 使用
三、信号
- 内置信号
- 自定义信号
四、多app应用
五、flask-script
- 安装
- 使用

一、数据库连接池

后端常常需要使用数据库的数据

当我们将数据库链接对象做成全局的就会产生，多个线程操作一个数据库链接对象时，容易导致数据错乱现象（例如一个线程使用该链接对象获取数据还未操作完毕，另一个线程使用该链接对象读取了其他数据，就产生了数据错乱）。

当我们链接数据库时时每个线程开一个链接，这样来了很多请求之后，链接过多会极大的消耗网络、内存资源导致系统运行速度过慢，甚至卡死。

所以针对于flask中我们使用第三方模块dbutils来管理数据库链接数量，这样每次线程执行数据库相关操作都只是从连接池中获取一个连接来进行，无论多少线程过来都只会轮流交替使用连接池的中连接。

DBUtils是Python的一个用于实现数据库连接池的模块

此连接池有两种连接模式：

DBUtils提供两种外部接口：

PersistentDB ：提供线程专用的数据库连接，并自动管理连接。
PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。

安装 dbutils

pip3 install dbutils

dbutils使用

PersistentDB模式：
为每个线程创建一个连接，线程即使调用了close方法，也不会关闭，只是把连接重新放到连接池，供自己线程再次使用。当线程终止时，连接自动关闭

PooledDB 模式：
创建一批连接到连接池，供所有线程共享使用。

示例以PooledDB 模式为例：

创建一个py文件，用来配置连接池
pool.py

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=[],  # 开始会话前执行的命令列表。
    ping=0,
    # ping MySQL服务端，检查是否服务可用。
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123',
    database='test',
    charset='utf8'
)

测试连接池代码

from flask import Flask
from pool import POOL

app = Flask(__name__)

app.debug = True

@app.route('/')
def show_db():
    # 从池中拿链接，创建出cursor对象
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute('select * from bbc')
    print(cursor.fetchall())
    return '查询成功'
if __name__ == '__main__':
    app.run()

二、wtfroms

wtfroms用于前后端混合的项目，与django中的forms类似
用于校验数据，渲染页面，显示提示信息

安装

pip3 install wtforms

使用

创建一个py文件编写wtforms
myform.py

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

app = Flask(__name__)
app.debug = True


class Myvalidators(object):
    '''自定义验证规则'''

    # 在__init__中将提示信息放入message中
    def __init__(self, message):
        self.message = message

    # 编写一些基础判断意外的验证规则
    def __call__(self, form, field):
        if field.data == "bbc":
            return None
        raise validators.ValidationError(self.message)

# 通过继承Form来设计字段验证规则
class LoginForm(Form):
    '''Form'''
    name = simple.StringField(
        label="用户名",
        widget=widgets.TextInput(),
        validators=[
            Myvalidators(message="用户名错误"),  # 也可以自定义正则
            validators.DataRequired(message="用户名不能为空"),
            validators.Length(max=8, min=3, message="用户名长度必须大于%(max)d且小于%(min)d")
        ],
        render_kw={"class": "form-control"}  # 设置展示在前端时的属性
    )

    pwd = simple.PasswordField(
        label="密码",
        validators=[
            validators.DataRequired(message="密码不能为空"),
            validators.Length(max=8, min=3, message="密码长度必须大于%(max)d且小于%(min)d"),
            validators.Regexp(regex="d+", message="密码必须是数字"),
        ],
        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("用户提交的数据用过格式验证，值为：%s" % form.data)
            return "登录成功"
        else:
            # print(form.errors, "错误信息")
            pass
        return render_template("login.html", form=form)


if __name__ == '__main__':
    app.run()

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>登录</title>
</head>
<body>
<form action="" method="post" novalidate>
    <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>

三、信号

内置信号

from flask import Flask, render_template
from flask import signals

app = Flask(__name__)


# 内置信号使用步骤

'''使用步骤
第一步：写个函数
第二步：跟信号绑定
第二步：跟信号绑定
'''


# 第一步：写个函数
def render_before(*args, **kwargs):
    print(args)
    print(kwargs)
    print('模板要渲染了')


# 第二步：跟信号绑定
'''
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在其中添加数据时，自动触发
'''


signals.before_render_template.connect(render_before)


# 第三步：触发信号 （内置信号，会自动触发）

@app.route('/')
def index():
    print('index 执行了')
    return render_template('index.html', name='刘亦菲')


if __name__ == '__main__':
    app.run()

自定义信号

from flask import Flask, render_template, request
from flask.signals import _signals

'''
第一步：定义信号
第二步：写个函数
第三步：跟信号绑定
第四步：触发信号
'''

app = Flask(__name__)

# 第一步：定义信号
xxx = _signals.signal('xxx')

# 第二步：写个函数
def add(*args, **kwargs):
    print(args)
    print(kwargs)
    print('add执行了')

# 第三步：跟信号绑定
xxx.connect(add)


# 第四步：触发信号

@app.route('/')
def index():
    xxx.send(request=request)  # 信号.send来触发设置好的自定义信号
    print('index 执行了')
    return render_template('index.html', name='bbc')


if __name__ == '__main__':
    app.run()

四、多app应用

# 一个flask项目，支持多个app对象
from flask import Flask
from werkzeug.middleware.dispatcher import DispatcherMiddleware
from werkzeug.serving import run_simple
app1 = Flask('app01')
app2 = Flask('app02')

@app1.route('/index')
def index():
    return "app01"

@app2.route('/index2')
def index2():
    return "app2"

dm = DispatcherMiddleware(app1, {
    '/sec': app2,
})

if __name__ == "__main__":
    # 请求来了，会执行 dm()---->触发DispatcherMiddleware的__call__
    run_simple('localhost', 5000, dm)

五、flask-script

想让flask能够执行指令来启动flask项目，可以自定制一些命令，来达成该效果

安装

pip3 install flask-script

使用

创建一个manage.py

from flask import Flask
# 使用步骤
from flask_script import Manager

app = Flask('app01')
# 基本使用，多了runserver命令
manager = Manager(app)


# 高级使用：自定制命令  python manage.py dbinit 

@manager.command
def dbinit():
    """
    python manage.py dbinit
    """
    print("数据库初始化完成")


@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 xxx
    执行： python manage.py  cmd --name lqz --url yyy
    """
    print(name, url)

@app.route('/index')
def index():
    return "app01"


if __name__ == "__main__":
    manager.run()

原文链接：https://blog.csdn.net/kdq18486588014/article/details/126348295