Flask中的数据连接池

pymsql链接数据库

import pymysql

conn = pymysql.connect(host='127.0.0.1',port=3306,user='root',passwd='123456',db='s8day127db')
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
# cursor.execute("select id,name from users where name=%s and pwd=%s",['lqz','123',])
cursor.execute("select id,name from users where name=%(user)s and pwd=%(pwd)s",{'user':'lqz','pwd':'123'})
obj = cursor.fetchone()
conn.commit()
cursor.close()
conn.close()

print(obj)

数据库连接池版

方式一:

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

from DBUtils.PersistentDB import PersistentDB
import pymysql
POOL = PersistentDB(
    creator=pymysql,# 使用链接数据库的模块
    maxusage=None,# 一个链接最多被重复使用的次数,None表示无限制
    setsession=[],# 开始会话前执行的命令列表。
    ping=0,# ping MySQL服务端,检查是否服务可用。
    closeable=False,# 如果为False时, conn.close() 实际上被忽略,供下次使用,再线程关闭时,才会自动关闭链接。如果为True时, conn.close()则关闭链接,那么再次调用pool.connection时就会报错,因为已经真的关闭了连接(pool.steady_connection()可以获取一个新的链接)
    threadlocal=None,# 本线程独享值得对象,用于保存链接对象,如果链接对象被重置
    host='127.0.0.1',password='123456',database='test',charset='utf8'
)

def func():
    conn = POOL.connection(shareable=False)
    cursor = conn.cursor()
    cursor.execute('select * from user')
    result = cursor.fetchall()
    print(result)
    cursor.close()
    conn.close()
if __name__ == '__main__':

    func()

方式二:

模式二:创建一批连接到连接池,供所有线程共享使用

setting.py

from datetime import timedelta
from redis import Redis
import pymysql
from DBUtils.PooledDB import PooledDB,SharedDBConnection

class Config(object):
    DEBUG = True
    SECRET_KEY = "umsuldfsdflskjdf"
    PERMANENT_SESSION_LIFETIME = timedelta(minutes=20)
    SESSION_REFRESH_EACH_REQUEST= True
    SESSION_TYPE = "redis"
    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',database='s8day127db',charset='utf8'
    )

class ProductionConfig(Config):
    SESSION_REDIS = Redis(host='192.168.0.94',port='6379')



class DevelopmentConfig(Config):
    SESSION_REDIS = Redis(host='127.0.0.1',port='6379')


class TestingConfig(Config):
    pass

utils/sql.py

import pymysql
from settings import Config
class SQLHelper(object):

    @staticmethod
    def open(cursor):
        POOL = Config.PYMYSQL_POOL
        conn = POOL.connection()
        cursor = conn.cursor(cursor=cursor)
        return conn,cursor

    @staticmethod
    def close(conn,cursor):
        conn.commit()
        cursor.close()
        conn.close()

    @classmethod
    def fetch_one(cls,sql,args,cursor =pymysql.cursors.DictCursor):
        conn,cursor = cls.open(cursor)
        cursor.execute(sql,args)
        obj = cursor.fetchone()
        cls.close(conn,cursor)
        return obj

    @classmethod
    def fetch_all(cls,args)
        obj = cursor.fetchall()
        cls.close(conn,cursor)
        return obj

使用:

obj = SQLHelper.fetch_one("select id,form.data)

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐


Jinja2:是Python的Web项目中被广泛应用的模板引擎,是由Python实现的模板语言,Jinja2 的作者也是 Flask 的作者。他的设计思想来源于Django的模板引擎,并扩展了其语法和一系列强大的功能,其是Flask内置的模板语言。
Fullcalendar日历使用,包括视图选择、事件插入、编辑事件、事件状态更改、事件添加和删除、事件拖动调整,自定义头部,加入el-popover显示图片、图片预览、添加附件链接等,支持手机显示。
监听QQ消息并不需要我们写代码,因为市面上已经有很多开源QQ机器人框架,在这里我们使用go-cqhttp官方文档:go-cqhttp如果您感兴趣的话,可以阅读一下官方文档,如果不想看,直接看我的文章即可。
【Flask框架】—— 视图和URL总结
python+web+flask轻量级框架的实战小项目。登录功能,后续功能可自行丰富。
有了这个就可以配置可信IP,关键是不需要企业认证,个人信息就可以做。
本专栏是对Flask官方文档中个人博客搭建进行的归纳总结,与官方文档结合事半功倍。 本人经验,学习一门语言或框架时,请首先阅读官方文档。学习完毕后,再看其他相关文章(如本系列文章),才是正确的学习道路。
本专栏是对Flask官方文档中个人博客搭建进行的归纳总结,与官方文档结合事半功倍。基础薄弱的同学请戳Flask官方文档教程 本人经验,学习一门语言或框架时,请首先阅读官方文档。学习完毕后,再看其他相关文章(如本系列文章),才是正确的学习道路。 如果python都完全不熟悉,一定不要着急学习框架,请首先学习python官方文档,一步一个脚印。要不然从入门到放弃是大概率事件。 Python 官方文档教程
快到年末了 相信大家都在忙着处理年末数据 刚好有一个是对超市的商品库存进行分析的学员案例 真的非常简单~
一个简易的问答系统就这样完成了,当然,这个项目还可以进一步完善,比如 将数据存入Elasticsearch,通过它先进行初步的检索,然后再通过这个系统,当然我们也可以用其他的架构实现。如果你对这系统还有其他的疑问,也可以再下面进行留言!!!
#模版继承和页面之间的调用@app.route("/bl")def bl(): return render_template("file_2.html")主ht
#form表达提交@app.route("/data",methods=['GET','POST']) #methods 让当前路由支持GET 和
#form表达提交@app.route("/data",methods=['GET','POST']) #methods 让当前路由支持GET 和
#session 使用app.secret_key = "dsada12212132dsad1232113"app.config['PERMANENT_SESSION_LI
#文件上传@app.route("/file",methods=['GET','POST'])def file(): if request.meth
#跳转操作:redirect@app.route("/red")def red(): return redirect("/login")
#session 使用app.secret_key = "dsada12212132dsad1232113"app.config['PERMANENT_SESSION_LI
@app.route("/req",methods=['GET','POST'])def req(): print(request.headers)
#模版继承和页面之间的调用@app.route("/bl")def bl(): return render_template("file_2.html")主ht
#文件操作:send_file,支持图片 视频 mp3 文本等@app.route("/img")def img(): return send_file("1.jpg&q