首页 > 数据库> 文章详细

数据库连接池，各种锁，信号，flask-script自定义命令

2022-08-10 21:35:58 阅读：208 来源： 互联网

标签：__ name 自定义 script flask app signals print

数据库连接池

flask 可以使用pymysql来操作数据库

借助于第三方模块，实现数据库连接池

1.安装 pip install dbutils

2.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='luffy_api',
    charset='utf8'
)
'''

使用

from flask import Flask
from pool import POOL

app = Flask(__name__)

@app.route('/banner')
def banner():
    # 从连接池中拿连接，创建cursor对象
    conn = POOL.connection()
    cursor = conn.cursor()
   	cursor.execute('select * from db')
    print(cursor.fetchall())
    return 'hello'

查看当前有多少个连接数

mysql终端：
	show status like 'Threads%'

锁

# 并发编程学的锁：
	-互斥锁：Lock
    -递归锁(可重入锁):RLock,互斥锁会出现死锁现象
    -Semaphore：多把锁,信号量
    -event事件：某个条件完成成，线程执行
# 所有锁:https://zhuanlan.zhihu.com/p/489305763

信号

django中也有信号，参考https://www.cnblogs.com/liuqingzheng/articles/9803403.html

安装

pip install blinker

概念

在整个flask请求生命周期中，运行到某一个特定的地方时，会自动触发（aop理念，面向切面编程的理念）

内置信号

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

内置信号使用步骤

from flask import Flask, render_template
from flask import signals

app = Flask(__name__)

(1)写一个函数
def render_brfore(*args,**kwargs):
	print(args)
    print(kwargs)
    print('模板要渲染了')
    
(2)跟信号绑定
signals.before_render_template.connect(render_before)


(3)触发信号
@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 import signals
from flask.signals import _signals

app = Flask(__name__)

(1)定义信号
xxx = _signals.signal('xxx')

(2)写个函数
def add(*args,**kwargs):
	print(args)
	print(kwargs)
	print('add执行了')
    
(3)绑定信号
xxx.connect(add)

(4)触发信号
@app.route('/')
def index():
    xxx.send()
    print('index 执行了')
    return render_template('index.html',name='春游去动物园')

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

flask-script

djagno中执行djagno程序，通过命令：python3 manage.py runserver--->还有其他很多命令
想让flask能够执行python3 manage.py runserver启动flask项目，自定制一些命令，完成更多操作


借助于第三方 flask-script来完成

pip install flask-script

from flask import Flask

from flask_script import Manager

app = Flask(__name__)

# 基本使用  
manager = Manager(app) # 使用Manager类将app包裹起来
'''
	现在只能使用命令方式运行项目，多了runserver方法
'''

# 高级使用：自定义命令 
@manager.command
def dbinit():
    '''
    	python manager = Manager(app)所在的文件名 dbinit
    	python manager.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__":
    # app.run()
    manager.run() # 需要修改为manager.run()

标签：__,name,自定义,script,flask,app,signals,print
来源： https://www.cnblogs.com/chunyouqudongwuyuan/p/16573928.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9