flask - 数据库连接池

分类：Flask作者：编程之家用户

flask

详细

  1. 编程语言及他们之间的区别？

    C/C++ ,很多语言的底层实现都是与C，代码执行效率高，自己做内存管理，对代码要求比较高，很多功能需要手动试下。
Java，比较好的编程语言，很多企业级应用都会选择java。
C#，是微软开始的编程语言，部署时需要放在windown server 上，最大弊端是window系统花钱。
PHP，一般用于快速搭建网站。
Python，简洁且入门简单，很少的代码就可以做很多事情。
Golang，语法和C比较接近，处理并发时比较有优势 + docker。

1.数据库连接池

1.1 初级阶段

import pymysql
from flask import Flask

app = Flask(__name__)


def fetchall(sql):
    conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
    cursor = conn.cursor()
    cursor.execute(sql)
    result = cursor.fetchall()
    cursor.close()
    conn.close()
    return result

@app.route('/login')
def login():
    result = fetchall('select * from user')
    return 'login'


@app.route('/index')
def index():
    result = fetchall('select * from user')
    return 'xxx'


@app.route('/order')
def order():
    result = fetchall('select * from user')
    return 'xxx'


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

1.2 老板给修改

import pymysql
from flask import Flask

app = Flask(__name__)

CONN = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')

def fetchall(sql):
    cursor = CONN.cursor()
    cursor.execute(sql)
    result = cursor.fetchall()
    cursor.close()
    return result

@app.route('/login')
def login():
    result = fetchall('select * from user')
    return 'login'


@app.route('/index')
def index():
    result = fetchall('select * from user')
    return 'xxx'


@app.route('/order')
def order():
    result = fetchall('select * from user')
    return 'xxx'


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

1.3 数据库连接池

安装

pip3 install dbutils
pip3 install pymysql

使用

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模块
    maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数
    mincached=2,  # 初始化时，链接池中至少创建的链接，0表示不创建
    blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
    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='222',
    database='cmdb',
    charset='utf8'
)

# 去连接池中获取一个连接
conn = POOL.connection()

cursor = conn.cursor()
cursor.execute('select * from web_models_disk')
result = cursor.fetchall()
cursor.close()

# 将连接放会到连接池
conn.close()

print(result)

多线程测试

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模块
    maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数
    mincached=2,  # 初始化时，链接池中至少创建的链接，0表示不创建
    blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
    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='222',
    database='cmdb',
    charset='utf8'
)


def task(num):
    # 去连接池中获取一个连接
    conn = POOL.connection()
    cursor = conn.cursor()
    # cursor.execute('select * from web_models_disk')
    cursor.execute('select sleep(3)')
    result = cursor.fetchall()
    cursor.close()
    # 将连接放会到连接池
    conn.close()
    print(num,'------------>',result)


from threading import Thread
for i in range(57):
    t = Thread(target=task,args=(i,))
    t.start()
基于函数实现sqlhelper
import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模块
    maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数
    mincached=2,  # 初始化时，链接池中至少创建的链接，0表示不创建
    blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
    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='222',
    database='cmdb',
    charset='utf8'
)

def fetchall(sql,*args):
    """ 获取所有数据 """
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute(sql,args)
    result = cursor.fetchall()
    cursor.close()
    conn.close()

    return result

def fetchone(sql, *args):
    """ 获取单挑数据 """
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute(sql, args)
    result = cursor.fetchone()
    cursor.close()
    conn.close()

    return result

from flask import Flask
import sqlhelper

app = Flask(__name__)

@app.route('/login')
def login():
    result = sqlhelper.fetchone('select * from web_models_admininfo where username=%s ','wupeiqi')
    print(result)
    return 'login'


@app.route('/index')
def index():
    result = sqlhelper.fetchall('select * from web_models_disk')
    print(result)
    return 'xxx'


@app.route('/order')
def order():
    # result = fetchall('select * from user')
    return 'xxx'


if __name__ == '__main__':
    app.run()
基于类实现sqlhelper
import pymysql
from DBUtils.PooledDB import PooledDB

class SqlHelper(object):
    def __init__(self):
        self.pool = PooledDB(
            creator=pymysql,  # 使用链接数据库的模块
            maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数
            mincached=2,  # 初始化时，链接池中至少创建的链接，0表示不创建
            blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
            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='222',
            database='cmdb',
            charset='utf8'
        )

    def open(self):
        conn = self.pool.connection()
        cursor = conn.cursor()
        return conn,cursor

    def close(self,cursor,conn):
        cursor.close()
        conn.close()

    def fetchall(self,sql, *args):
        """ 获取所有数据 """
        conn,cursor = self.open()
        cursor.execute(sql, args)
        result = cursor.fetchall()
        self.close(conn,cursor)
        return result

    def fetchone(self,sql, *args):
        """ 获取所有数据 """
        conn, cursor = self.open()
        cursor.execute(sql, args)
        result = cursor.fetchone()
        self.close(conn, cursor)
        return result


db = SqlHelper()

from flask import Flask
from sqlhelper2 import db

app = Flask(__name__)

@app.route('/login')
def login():
    # db.fetchone()
    return 'login'


@app.route('/index')
def index():
    # db.fetchall()
    return 'xxx'

@app.route('/order')
def order():
    # db.fetchall()
    conn,cursor = db.open()
    # 自己做操作
    db.close(conn,cursor)
    return 'xxx'

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

2.补充知识点：上下文管理

class Foo(object):

    def __enter__(self):
        return 123

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

obj = Foo()

with obj as f:
    print(f)

class Foo(object):

    def do_somthing(self):
        pass

    def close(self):
        pass
    
class Context:
    def __enter__(self):
        self.data = Foo()
        return self.data

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.data.close()

with Context() as ctx:
    ctx.do_somthing()

原文地址：https://www.cnblogs.com/zhaoganggang/p/13445429.html

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

相关推荐

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