526互联

连接池、信号、flask-script

发布时间 2023-08-22 20:14:35作者: 星空看海

目录

一 数据库连接池

# flask--->数据库--->原生操作--->pymsql

1.1 数据库连接池

# 1 安装：DBUtils
# 2 使用 类创建一个池对象
PYMYSQL_POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模块
    maxconnections=2,  # 连接池允许的最大连接数，0和None表示不限制连接数
    mincached=1,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
    maxcached=0,  # 链接池中最多闲置的链接，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='123',
    database='cars',
    charset='utf8'
)
# 3 从池对象中，取出一个链接使用
	conn = PYMYSQL_POOL.connection()

    
    
# 4 flask中使用
@app.route('/')
def index():
    conn = PYMYSQL_POOL.connection()  # 从池中拿一个链接
    cursor = conn.cursor(cursor=DictCursor)  # 默认元组套元组，设置DictCursor就是列表套字典
    cursor.execute('select id,title from news where id<10')
    res1 = cursor.fetchall()
    cursor.close()
    conn.close()
    return jsonify(res1)

Django连接数据池

https://www.cnblogs.com/wangruixing/p/13030755.html

二 信号

# Flask框架中的信号基于blinker，其主要就是让开发者可是在flask请求过程中定制一些用户行为


# 内置信号
# 自定义信号

2.1 内置信号的使用

Flask框架中的信号基于blinker，其主要就是让开发者可是在flask请求过程中定制一些用户行为

# 安装：pip3 install blinker

from flask import Flask, render_template, signals

app = Flask(__name__)
app.debug = True


# 1 使用内置信号实现，每次模板渲染之前，都执行某个代码

# 第一步：写一个函数
def test(app, **kwargs):
    print(app)
    print(type(kwargs))
    # 请求地址是根路径，才记录日志，其它都不记录
    print(kwargs['context']['request'].path)
    if kwargs['context']['request'].path == '/':
        print('记录日志了')


# 第二步：跟内置信号绑定
# signals 中有很多内置信号
signals.before_render_template.connect(test)


# 第三步：等待信号被触发(不需要咱们做)--->只要执行到内置信号位置，绑定的函数就会执行

@app.route('/')
def index():
    return render_template('index.html')


@app.route('/home')
def home():
    return render_template('home.html')


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

2.2 有哪些内置信号

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在其中添加数据时，自动触发

2.3 自定义信号

# 1 第一步：定义一个自定义 信号
# 2 第二步：写个函数
# 3 第三步：函数跟自己定义信号绑定
# 4 第四步：触发自定义信号--->我们做

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

app = Flask(__name__)
app.debug = True

#### 自定义信号---> session每次放一个值，我们就执行信号
# 1 定义信号
session_input = _signals.signal('session_input')


# 2 写个函数
def test2(*args, **kwargs):
    print(args)  # app
    print(kwargs)  # {session   kk}
    print('session放值了')
    if kwargs.get('kk').get('name') == 'xx':
        print('记录日志')


# 3 绑定信号
session_input.connect(test2)


@app.route('/')
def index():
    session['uu'] = '00'
    # 4 触发信号
    session_input.send(app, session=session, kk={'name': 'uu'})
    return render_template('index.html')


@app.route('/home')
def home():
    return render_template('home.html')


@app.route('/order')
def order():
    session['xx'] = 'xx'
    # 4 触发信号
    session_input.send(app, session=session, kk={'name': 'xx'})
    return "order"


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

2.4 信号的作用（信号量-->Semaphore）

# 对代码进行解耦  

# 1 记录日志：只要是张三用户，访问index页面，就记录日志
# 2 只要用户表中，插入一条记录，就给用户发个短信通知
	User.object.create--->调用发短信方法-->找到10个地址-->改10个地方
    -如果有个内置信号--->只要表中增加记录，就会触发这个信号---->通过信号内判断这个表是不是User表，决定要不要发短信
    -flask中没有这个内置信号--->自定义

2.5 补充：信号量(线程中)

# acquire()：消耗信号量
# release()：释放信号量
import threading
import time
def run(n):
    semaphore.acquire()  # 计数器获取锁
    time.sleep(5)  # 程序休眠5秒
    print (n)
    semaphore.release()  # 计数器释放锁


if __name__ == '__main__':
    # 添加一个计数器，最大并发线程数量5（最多同时运行5个线程）
    semaphore = threading.Semaphore(5)
    for i in range(50):
        t = threading.Thread(target=run, args=(i, ))  # 创建线程
        t.start()

三 django的信号

# 内置信号：
	#Model signals
    pre_init                    # django的modal执行其构造方法前，自动触发
    post_init                   # django的modal执行其构造方法后，自动触发
    pre_save                    # django的modal对象保存前，自动触发
    post_save                   # django的modal对象保存后，自动触发
    pre_delete                  # django的modal对象删除前，自动触发
    post_delete                 # django的modal对象删除后，自动触发
    m2m_changed                 # django的modal中使用m2m字段操作第三张表（add,remove,clear）前后，自动触发
    class_prepared              # 程序启动时，检测已注册的app中modal类，对于每一个类，自动触发
Management signals
    pre_migrate                 # 执行migrate命令前，自动触发
    post_migrate                # 执行migrate命令后，自动触发
Request/response signals
    request_started             # 请求到来前，自动触发
    request_finished            # 请求结束后，自动触发
    got_request_exception       # 请求异常后，自动触发
Test signals
    setting_changed             # 使用test测试修改配置文件时，自动触发
    template_rendered           # 使用test测试渲染模板时，自动触发
Database Wrappers
    connection_created          # 创建数据库连接时，自动触发

3.1 内置信号的使用

# 内置信号使用(当user表创建用户，就给用户发个邮件)
	1 写个函数   #放到__init__里
        from django.db.models.signals import pre_save
        import logging
        def callBack(sender, **kwargs):
            logging.debug('%s创建了一个%s对象'%(sender._meta.model_name,kwargs.get('instance').title))

        
     2 绑定内置信号   
		pre_save.connect(callBack)
	3 等待触发
    
    
 # 内置信号
    from django.db.models.signals import pre_save
    from django.dispatch import receiver
    @receiver(pre_save)
    def my_callback(sender, **kwargs):
        print("对象创建成功")
        print(sender)
        print(kwargs)

3.2 自定义信号

	#1 定义信号（一般创建一个py文件）（toppings，size 是接受的参数）

    import django.dispatch
    pizza_done = django.dispatch.Signal(providing_args=["toppings", "size"])
    # 2 注册信号
    def callback(sender, **kwargs):
        print("callback")
        print(sender,kwargs)
    pizza_done.connect(callback)
# 3 触发信号
    from 路径 import pizza_done
    pizza_done.send(sender='seven',toppings=123, size=456)
    
    
    
# 自定义信号干过什么？
	-做双写一致性的缓存更新
    https://www.cnblogs.com/zjyao/p/17527227.html

四 flask-script

# django中，有命令  
	python manage.py runserver
    python manage.py makemigrations
    ...自定制命令（django如何自定制命令）...
    	-python manage.py init_db  excel文件路径  指定表名


# flask启动项目，像djagno一样，通过命令启动

Flask==2.2.2
Flask_Script==2.0.3


#借助于：flask-script 实现
	-安装：pip3.8 install flask-script
    -修改代码：
    	from flask_script import Manager
    	manager=Manager(app)
        manager.run()
    -用命令启动
    	python38 manage.py runserver

4.1 自定制命令

# 自定制命令
    #1  简单自定制命令
    @manager.command
    def custom(arg):
        # 命令的代码，比如：初始化数据库， 有个excel表格，使用命令导入到mysql中
        print(arg)

    #2 复杂一些的自定制命令
    @manager.option('-n', '--name', dest='name')
    @manager.option('-u', '--url', dest='url')
    def cmd(name, url):
        # python run.py cmd -n lqz -u xxx
        # python run.py cmd --name lqz --url uuu
        print(name, url)
    
    

# django 中如何自定制命令
https://blog.csdn.net/zhouruifu2015/article/details/129761804