花生_TL007

这个屌丝很懒，什么也没留下！

热门标签

Flask 数据库连接池、DBUtils、http 连接池_flask mysql 连接池

作者：花生_TL007 | 2024-04-13 03:46:23

踩

flask mysql 连接池

1、DBUtils 简介、使用

DBUtils 简介

DBUtils 是一套用于管理数据库 "连接池" 的Python包，为 "高频度、高并发" 的数据库访问提供更好的性能，可以自动管理连接对象的创建和释放。并允许对非线程安全的数据库接口进行线程安全包装和连接。该连接可在各种多线程环境中使用。

使用场景：如果使用的是流行的对象关系映射器 SQLObject 或 SQLAlchemy 之一，则不需要 DBUtils，因为它们带有自己的连接池。SQLObject 2 (SQL-API) 实际上是从 DBUtils 中借用了一些代码，将池化分离到一个单独的层中。

DBUtils 提供两种外部接口：

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

另外，实际使用的数据库驱动也有所依赖，比如SQLite数据库只能使用PersistentDB作连接池。下载地址：http://www.webwareforpython.org/downloads/DBUtils/

使用 DBUtils 数据库连接池

安装：pip install DBUtils

示例：MySQLdb 模块使用

连接池对象只初始化一次，一般可以作为模块级代码来确保。 PersistentDB 的连接例子:


import DBUtils.PersistentDB
 
# maxusage 则为一个连接最大使用次数
persist = DBUtils.PersistentDB.PersistentDB(dbpai=MySQLdb,maxusage=1000,**kwargs)
 
# 获取连接池
conn = persist.connection() 
 
# 关闭连接池
conn.close()

参数 dbpai 指定使用的数据库模块，兼容 DB-API 。下面是支持 DB-API 2 规范的数据库模块

pip install pymysql（mysql）
pip install pymssql（sqlserver）
pip install cx_Oracle（oracle）
pip install phoenixdb（hbase）
pip install sqlite3（sqlite3 python自带）

DBUtils 仅提供给了连接池管理，实际的数据库操作依然是由符合 DB-API 2 标准的目标数据库模块完成的。

示例：pymysql 模块使用

PooledDB 使用方法同 PersistentDB，只是参数有所不同。

dbapi ：数据库接口
mincached ：启动时开启的空连接数量
maxcached ：连接池最大可用连接数量
maxshared ：连接池最大可共享连接数量
maxconnections ：最大允许连接数量
blocking ：达到最大数量时是否阻塞
maxusage ：单个连接最大复用次数
setsession ：用于传递到数据库的准备会话，如 [”set name UTF-8″] 。

conn = pooled.connection()
cur = conn.cursor()
cur.execute(sql)
res = cur.fetchone()
cur.close() # 或者 del cur
conn.close() # 或者 del conn


import pymysql
from dbutils.pooled_db import PooledDB
 
# 定义连接参数
pool = PooledDB(
    creator=pymysql,
    maxconnections=6,
    mincached=2,
    maxcached=5,
    blocking=True,
    host='localhost',
    user='root',
    passwd='123456',
    db='mydb',
    port=3306,
    charset='utf8mb4'
)
 
 
def main():
    # 从连接池获取连接
    conn = pool.connection()
    cursor = conn.cursor()
 
    # 执行 SQL 语句
    sql = "SELECT * FROM students"
    cursor.execute(sql)
    result = cursor.fetchall()
 
    # 处理查询结果
    for row in result:
        print(row)
 
    # 关闭游标和连接
    cursor.close()
    conn.close()
 
 
if __name__ == '__main__':
    main()

示例：面向对象使用 DBUtils


"""
使用DBUtils数据库连接池中的连接，操作数据库
"""
import json
import pymysql
import datetime
from DBUtils.PooledDB import PooledDB
import pymysql
 
 
class MysqlClient(object):
    __pool = None;
 
    def __init__(self, mincached=10, maxcached=20, maxshared=10, maxconnections=200, blocking=True,
                 maxusage=100, setsession=None, reset=True,
                 host='127.0.0.1', port=3306, db='test',
                 user='root', passwd='123456', charset='utf8mb4'):
        """
        :param mincached:连接池中空闲连接的初始数量
        :param maxcached:连接池中空闲连接的最大数量
        :param maxshared:共享连接的最大数量
        :param maxconnections:创建连接池的最大数量
        :param blocking:超过最大连接数量时候的表现，为True等待连接数量下降，为false直接报错处理
        :param maxusage:单个连接的最大重复使用次数
        :param setsession:optional list of SQL commands that may serve to prepare
            the session, e.g. ["set datestyle to ...", "set time zone ..."]
        :param reset:how connections should be reset when returned to the pool
            (False or None to rollback transcations started with begin(),
            True to always issue a rollback for safety's sake)
        :param host:数据库ip地址
        :param port:数据库端口
        :param db:库名
        :param user:用户名
        :param passwd:密码
        :param charset:字符编码
        """
 
        if not self.__pool:
            self.__class__.__pool = PooledDB(pymysql,
                                             mincached, maxcached,
                                             maxshared, maxconnections, blocking,
                                             maxusage, setsession, reset,
                                             host=host, port=port, db=db,
                                             user=user, passwd=passwd,
                                             charset=charset,
                                             cursorclass=pymysql.cursors.DictCursor
                                             )
        self._conn = None
        self._cursor = None
        self.__get_conn()
 
    def __get_conn(self):
        self._conn = self.__pool.connection();
        self._cursor = self._conn.cursor();
 
    def close(self):
        try:
            self._cursor.close()
            self._conn.close()
        except Exception as e:
            print(e)
 
    def __execute(self, sql, param=()):
        count = self._cursor.execute(sql, param)
        print(count)
        return count
 
    @staticmethod
    def __dict_datetime_obj_to_str(result_dict):
        """把字典里面的datatime对象转成字符串，使json转换不出错"""
        if result_dict:
            result_replace = {k: v.__str__() for k, v in result_dict.items() if isinstance(v, datetime.datetime)}
            result_dict.update(result_replace)
        return result_dict
 
    def select_one(self, sql, param=()):
        """查询单个结果"""
        count = self.__execute(sql, param)
        result = self._cursor.fetchone()
        """:type result:dict"""
        result = self.__dict_datetime_obj_to_str(result)
        return count, result
 
    def select_many(self, sql, param=()):
        """
        查询多个结果
        :param sql: qsl语句
        :param param: sql参数
        :return: 结果数量和查询结果集
        """
        count = self.__execute(sql, param)
        result = self._cursor.fetchall()
        """:type result:list"""
        [self.__dict_datetime_obj_to_str(row_dict) for row_dict in result]
        return count, result
 
    def execute(self, sql, param=()):
        count = self.__execute(sql, param)
        return count
 
    def begin(self):
        """开启事务"""
        self._conn.autocommit(0)
 
    def end(self, option='commit'):
        """结束事务"""
        if option == 'commit':
            self._conn.autocommit()
        else:
            self._conn.rollback()
 
 
if __name__ == "__main__":
    mc = MysqlClient()
    sql1 = 'SELECT * FROM shiji  WHERE  id = 1'
    result1 = mc.select_one(sql1)
    print(json.dumps(result1[1], ensure_ascii=False))
 
    sql2 = 'SELECT * FROM shiji  WHERE  id IN (%s,%s,%s)'
    param = (2, 3, 4)
    print(json.dumps(mc.select_many(sql2, param)[1], ensure_ascii=False))

不用连接池


import MySQLdb
conn= MySQLdb.connect(host='localhost',user='root',passwd='pwd',db='myDB',port=3306)  
#import pymysql
#conn = pymysql.connect(host='localhost', port='3306', db='game', user='root', password='123456', charset='utf8')
cur=conn.cursor()
SQL="select * from table1"
r=cur.execute(SQL)
r=cur.fetchall()
cur.close()
conn.close()

使用连接池


import MySQLdb
from DBUtils.PooledDB import PooledDB
#5为连接池里的最少连接数
pool = PooledDB(MySQLdb,5,host='localhost',user='root',passwd='pwd',db='myDB',port=3306) 
 
# 以后每次需要数据库连接就是用connection（）函数获取连接就好了
conn = pool.connection()  
cur=conn.cursor()
SQL="select * from table1"
r=cur.execute(SQL)
r=cur.fetchall()
cur.close()
conn.close()

多线程使用连接池


import sys
import threading
import MySQLdb
import DBUtils.PooledDB
 
connargs = { "host":"localhost", "user":"user1", "passwd":"123456", "db":"test" }
def test(conn):
    try:
        cursor = conn.cursor()
        count = cursor.execute("select * from users")
        rows = cursor.fetchall()
        for r in rows: pass
    finally:
        conn.close()
        
def testloop():
    print ("testloop")
    for i in range(1000):
        conn = MySQLdb.connect(**connargs)
        test(conn)
        
def testpool():
    print ("testpool")
    pooled = DBUtils.PooledDB.PooledDB(MySQLdb, **connargs)
    for i in range(1000):
        conn = pooled.connection()
        test(conn)
        
def main():
    t = testloop if len(sys.argv) == 1 else testpool
    for i in range(10):
        threading.Thread(target = t).start()
        
if __name__ == "__main__":
    main()

虽然测试方式不是很严谨，但从测试结果还是能感受到 DBUtils 带来的性能提升。当然，我们我们也可以在 testloop() 中一直重复使用一个不关闭的 Connection，但这却不适合实际开发时的情形。

2、Flask 配置，蓝图，数据库连接池，上下文原理

Flask之配置文件，蓝图，数据库连接池，上下文原理：https://www.cnblogs.com/yunweixiaoxuesheng/p/8418135.html

配置

方式一，使用字典方式配置

app.config['SESSION_COOKE_NAME'] = 'session_liling'

方式二，引入文件，设置

from flask import Flask

app = Flask(__name__)

app.config.from_pyfile('settings.py') # 引用settings.py中的AAAA
print(app.config['AAAA']) # 123

# settings.py
AAAA = 123

方法三，使用环境变量设置，推荐使用

from flask import Flask

app = Flask(__name__)

import os

os.environ['FLASK-SETTINGS'] = 'settings.py'

app.config.from_envvar('FLASK-SETTINGS')

方式四，通过对象方式导入使用,可根据不同环境选择不同的配置，推荐使用

from flask import Flask

app = Flask(__name__)

app.config.from_object('settings.BaseConfig')
print(app.config['NNNN']) # 123

# settings.py

class BaseConfig(object): # 公用配置
NNNN = 123

class TestConfig(object):
DB = '127.0.0.1'

class DevConfig(object):
DB = '192.168.1.1'

class ProConfig(object):
DB = '47.18.1.1'

不同的文件引用配置


from flask import Flask,current_app
 
app = Flask(__name__)
 
app.secret_key = 'adfadsfhjkhakljsdfh'
 
app.config.from_object('settings.BaseConfig')
 
@app.route('/index')
def index():
    print(current_app.config['NNNN'])
    return 'xxx'
 
if __name__ == '__main__':
    app.run()

instance_path、instance_relative_config


from flask import Flask,current_app
 
app = Flask(__name__,instance_path=None,instance_relative_config=False)
# 默认 instance_relative_config = False 那么 instance_relative_config和instance_path都不会生效
# instance_relative_config=True，instance_path才会生效，app.config.from_pyfile('settings.py')将会失效
# 配置文件找的路径，按instance_path的值作为配置文件路径
# 默认instance_path=None，None会按照当前路径下的instance文件夹为配置文件的路径
# 如果设置路径，按照设置的路径查找配置文件。
 
app.config.from_pyfile('settings.py')
 
@app.route('/index')
def index():
    print(current_app.config['NNNN'])
    return 'xxx'
 
 
if __name__ == '__main__':
    app.run()

蓝图

对应用程序的目录结构进行分配，一般适用于小中型企业

代码：


# crm/__init__.py
# 创建flask项目，用蓝图注册不同的模块
 
from flask import Flask
from .views import account
from .views import order
 
app = Flask(__name__)
 
app.register_blueprint(account.account)
app.register_blueprint(order.order)
 
------------------------------------------------------------------------
# manage.py
# 启动文件
import crm
 
if __name__ == '__main__':
    crm.app.run()
 
------------------------------------------------------------------------
# crm/views/account.py
# 视图函数模块,Blueprint，将函数引入app
from flask import Blueprint
 
account = Blueprint('account',__name__,url_prefix='/xxx')
 
@account.route('/login')
def login():
    return 'Login'

Flask 数据库连接池

：https://www.cnblogs.com/TheLand/p/9178305.html

ORM ( 对象关系映射 )

ORM（Object-Relational Mapping，对象关系映射）是一种编程技术，用于在关系型数据库和面向对象编程语言之间建立映射关系。它允许开发人员使用面向对象的方式来操作数据库，而无需直接编写或执行 SQL 查询。

ORM 提供了一个抽象层，将数据库表格映射为对象，并提供了一组方法和工具，以便于进行数据库的增删改查操作。开发人员可以通过使用对象和方法来表示和操作数据，而不必关心底层的 SQL 语句和数据库细节。

常见的 ORM 框架包括：

SQLAlchemy：是 python 操作数据库的一个库，能够进行 orm 映射，是一个功能强大的 Python ORM 框架，支持多种数据库后端，提供了高级的查询功能和事务管理等特性。是为高效和高性能的数据库访问设计，实现了完整的企业级持久模型。SQLAlchemy 的理念是，SQL 数据库的量级和性能重要于对象集合；而对象集合的抽象又重要于表和行。
Flask-SQLAlchemy：Flask-SQLAlchemy 是一个与 Flask 框架集成的 SQLAlchemy 扩展，它简化了在 Flask 应用程序中使用 SQLAlchemy 进行数据库操作的过程。它提供了一组简单而强大的工具和功能，使得与数据库的交互变得更加轻松和高效。
Django ORM：Django 框架自带的 ORM，提供了简单易用的接口，支持多种数据库后端，并具有强大的查询和模型关联功能。
Hibernate：Java 领域中最流行的 ORM 框架，为 Java 对象和关系型数据库之间提供了映射和管理。

使用 ORM 的好处包括：

提高开发效率：ORM 提供了面向对象的编程接口，使得开发人员能够更快速地进行数据库操作，减少了编写和调试 SQL 语句的工作量。
跨数据库平台：ORM 框架通常支持多种数据库后端，使得开发人员能够轻松地切换或同时使用不同的数据库系统。
数据库抽象和安全性：ORM 隐藏了底层的数据库细节，提供了一层抽象，有助于维护和管理数据库结构，并提供了安全性保护，如参数绑定和防止 SQL 注入。
更好的可维护性和可测试性：使用 ORM 可以提高代码的可读性和可维护性，使得进行单元测试和集成测试更加容易。

注意：ORM 并不能解决所有数据库问题。在某些情况下，复杂的查询和性能要求可能需要直接使用原生 SQL。因此，根据具体的需求和场景，谨慎选择和使用合适的 ORM 框架。

为什么使用数据库连接池

多连接：如果不用连接池时，每次操作都要链接数据库，链接次数过多，数据库会耗费过多资源，数量过大的话，数据库会过载，导致程序运行缓慢。
单连接：在程序中全局创建连接，导致程序会一直使用一个连接，避免了反复连接造成的问题。但是多线程时就得加锁。这样就变成串行，没法实现并发

解决方法：

方式 1：为每一个线程创建一个链接（是基于本地线程来实现的。thread.local），每个线程独立使用自己的数据库链接，该线程关闭不是真正的关闭，本线程再次调用时，还是使用的最开始创建的链接，直到线程终止，数据库链接才关闭。如果线程比较多还是会创建很多连接
方式 2：创建一个链接池，为所有线程提供连接，使用时来进行获取，使用完毕后在放回到连接池。假设最大链接数有10个，其实也就是一个列表，当你pop一个，人家会在append一个，链接池的所有的链接都是按照排队的这样的方式来链接的。链接池里所有的链接都能重复使用，共享的，即实现了并发，又防止了链接次数太多

基于 DBUtils 数据库连接池

数据库连接池避免每次操作都要连接数据库，一直使用一个连接，多线程也会出现问题，可加锁，但变为串行


import pymysql
import threading
from threading import RLock
 
LOCK = RLock()
CONN = pymysql.connect(
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123',
    database='ok1',
    charset='utf8'
)
 
 
def task(arg):
    with LOCK:
        cursor = CONN.cursor()
        cursor.execute('select * from book')
        result = cursor.fetchall()
        cursor.close()
        print(result)
 
 
for i in range(10):
    t = threading.Thread(target=task, args=(i,))
    t.start()

线程之间的数据隔离

"本地线程" 可以实现线程之间的数据隔离。保证每个线程都只有自己的一份数据，在操作时不会影响别人的，即使是多线程，自己的值也是互相隔离的


import threading
import time
 
# 本地线程对象
local_values = threading.local()
 
 
def func(num):
    """
    # 第一个线程进来，本地线程对象会为他创建一个
    # 第二个线程进来，本地线程对象会为他创建一个
    {
        线程1的唯一标识：{name:1},
        线程2的唯一标识：{name:2},
    }
    :param num: 
    :return: 
    """
    local_values.name = num  # 4
    # 线程停下来了
    time.sleep(2)
    # 第二个线程： local_values.name，去local_values中根据自己的唯一标识作为key，获取value中name对应的值
    print(local_values.name, threading.current_thread().name)
 
 
for i in range(5):
    th = threading.Thread(target=func, args=(i,), name='线程%s' % i)
    th.start()

模式一：每个线程创建一个连接

基于threading.local实现创建每个连接。
每个线程会创建一个连接，该线程没有真正关闭。
再次调用该线程时，还是使用原有的连接。
线程真正终止的时候，连接才会关闭。


from DBUtils.PersistentDB import PersistentDB
import pymysql
 
POOL = PersistentDB(
    creator=pymysql,  # 使用链接数据库的模块
    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
    closeable=False,
    # 如果为False时， conn.close() 实际上被忽略，供下次使用，再线程关闭时，才会自动关闭链接。如果为True时， conn.close()则关闭链接，那么再次调用pool.connection时就会报错，因为已经真的关闭了连接（pool.steady_connection()可以获取一个新的链接）
    threadlocal=None,  # 本线程独享值得对象，用于保存链接对象，如果链接对象被重置
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123',
    database='pooldb',
    charset='utf8'
)
 
 
def func():
    # conn = SteadyDBConnection()
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    cursor.close()
    conn.close() # 不是真的关闭，而是假的关闭。 conn = pymysql.connect()   conn.close()
 
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    cursor.close()
    conn.close()
 
import threading
 
for i in range(10):
    t = threading.Thread(target=func)
    t.start()

模式二：线程复用连接池 (推荐)

创建一个连接池，为所有线程提供连接，线程使用连接时获取连接，使用完毕放回连接池。
线程不断地重用连接池里的连接。


import time
import pymysql
import threading
from DBUtils.PooledDB import PooledDB, SharedDBConnection
 
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 MySQL服务端，检查是否服务可用。
    # 如：0 = None = never, 1 = default = whenever it is requested, 
    # 2 = when a cursor is created, 4 = when a query is executed, 7 = always
    ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123456',
    database='flask_test',
    charset='utf8'
)
 
 
def func():
    # 检测当前正在运行连接数的是否小于最大链接数，如果不小于则：等待或报raise TooManyConnections异常
    # 否则
    # 则优先去初始化时创建的链接中获取链接 SteadyDBConnection。
    # 然后将SteadyDBConnection对象封装到PooledDedicatedDBConnection中并返回。
    # 如果最开始创建的链接没有链接，则去创建一个SteadyDBConnection对象，
    # 再封装到PooledDedicatedDBConnection中并返回。
    # 一旦关闭链接后，连接就返回到连接池让后续线程继续使用。
 
    # PooledDedicatedDBConnection
    conn = POOL.connection()
 
    # print(th, '链接被拿走了', conn1._con)
    # print(th, '池子里目前有', pool._idle_cache, '\r\n')
 
    cursor = conn.cursor()
    cursor.execute('select * from userinfo')
    result = cursor.fetchall()
    print(result)
    conn.close()
 
    conn = POOL.connection()
 
    # print(th, '链接被拿走了', conn1._con)
    # print(th, '池子里目前有', pool._idle_cache, '\r\n')
 
    cursor = conn.cursor()
    cursor.execute('select * from userinfo')
    result = cursor.fetchall()
    conn.close()
 
 
func()

上下文管理

所谓上下文，像考试题目根据上下文，回答一下问题。
程序中，泛指的外部环境，像wsgi来的网络请求，而且通常只有上文。
flask中的上下文，被使用在 current_app，session，request 上。

flask 本地线程


from flask import session
 
try:
    from greenlet import getcurrent as get_ident  # grenlet协程模块
except ImportError:
    try:
        from thread import get_ident
    except ImportError:
        from _thread import get_ident  # get_ident()，获取线程的唯一标识
 
 
class Local(object):  # 引用session中的LocalStack下的Local
    __slots__ = ('__storage__', '__ident_func__')  # __slots__该类在外面调用时，只能调用定义的字段，其他的不能调用
 
    def __init__(self):
        # object.__setattr__为self设置值，等价于self.__storage__ = {}
        # 为父类object中包含的__steattr__方法中的self.__storage__ = {}
        # 由于类内包含__steattr__，self.xxx（对象.xxx）时会自动会触发__steattr__，
        # 当前__steattr__中storage = self.__storage__又会像self.xxx要值，故会造成递归
        # 所以在父类中__steattr__方法赋值，避免self.xxx调用__setattr__造成的递归
        object.__setattr__(self, '__storage__', {})
 
        object.__setattr__(self, '__ident_func__', get_ident)  # 赋值为协程
 
    def __iter__(self):
        return iter(self.__storage__.items())
 
    def __release_local__(self):
        self.__storage__.pop(self.__ident_func__(), None)
 
    def __getattr__(self, name):
        try:
            return self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)
 
    def __setattr__(self, name, value):
        ident = self.__ident_func__()  # 获取单钱线程(协程)的唯一标识
        storage = self.__storage__  # {}
        try:
            storage[ident][name] = value  # { 111 : {'stack':[] },222 : {'stack':[] } }
        except KeyError:
            storage[ident] = {name: value}
 
    def __delattr__(self, name):
        try:
            del self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)
 
 
_local = Local()  # flask的本地线程功能，类似于本地线程，如果有人创建Local对象并，设置值，每个线程里一份
_local.stack = []  # _local.stack会调用__setattr__的self.__ident_func__()取唯一标识等

特殊栈


from flask import session
 
try:
    from greenlet import getcurrent as get_ident
except ImportError:
    try:
        from thread import get_ident
    except ImportError:
        from _thread import get_ident  # 获取线程的唯一标识 get_ident()
 
 
class Local(object):
    __slots__ = ('__storage__', '__ident_func__')
 
    def __init__(self):
        # self.__storage__ = {}
        # self.__ident_func__ = get_ident
        object.__setattr__(self, '__storage__', {})
        object.__setattr__(self, '__ident_func__', get_ident)
 
    def __iter__(self):
        return iter(self.__storage__.items())
 
    def __release_local__(self):
        self.__storage__.pop(self.__ident_func__(), None)
 
    def __getattr__(self, name):
        try:
            return self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)
 
    def __setattr__(self, name, value):
        ident = self.__ident_func__()  # 获取当前线程(协程)的唯一标识
        storage = self.__storage__  # {}
        try:
            storage[ident][name] = value  # { 111:{'stack':[] },222:{'stack':[] }  }
        except KeyError:
            storage[ident] = {name: value}
 
    def __delattr__(self, name):
        try:
            del self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)
 
 
_local = Local()
_local.stack = []

使用 flask 中的 stack 和 local


from functools import partial
from flask.globals import LocalStack, LocalProxy
 
_request_ctx_stack = LocalStack()
 
 
class RequestContext(object):
    def __init__(self, environ):
        self.request = environ
 
 
def _lookup_req_object(name):
    top = _request_ctx_stack.top
    if top is None:
        raise RuntimeError(_request_ctx_stack)
    return getattr(top, name)
 
 
# 实例化了LocalProxy对象，_lookup_req_object参数传递
session = LocalProxy(partial(_lookup_req_object, 'session'))
 
"""
local = {
    “标识”: {'stack': [RequestContext(),]}
}
"""
_request_ctx_stack.push(RequestContext('c1'))  # 当请求进来时，放入
 
print(session)  # 获取 RequestContext('c1')， top方法
print(session)  # 获取 RequestContext('c1')， top方法
_request_ctx_stack.pop()  # 请求结束pop

示例：


from functools import partial
from flask.globals import LocalStack, LocalProxy
 
ls = LocalStack()
 
 
class RequestContext(object):
    def __init__(self, environ):
        self.request = environ
 
 
def _lookup_req_object(name):
    top = ls.top
    if top is None:
        raise RuntimeError(ls)
    return getattr(top, name)
 
 
session = LocalProxy(partial(_lookup_req_object, 'request'))
 
ls.push(RequestContext('c1')) # 当请求进来时，放入
print(session) # 视图函数使用
print(session) # 视图函数使用
ls.pop() # 请求结束pop
 
 
ls.push(RequestContext('c2'))
print(session)
 
ls.push(RequestContext('c3'))
print(session)

Flask SQLAlchemy 使用连接池

Flask SQLAlchemy 提供了内置的连接池功能，可以方便地配置和使用。

在Flask应用中，我们可以通过配置 SQLALCHEMY_POOL_SIZE 参数来设置连接池的大小。连接池的大小决定了同时打开的数据库连接的数量。例如，我们可以将连接池的大小设置为10：

app.config['SQLALCHEMY_POOL_SIZE'] = 10


from flask_sqlalchemy import SQLAlchemy
 
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://user:password@localhost/db_name'
db = SQLAlchemy(app)
 
# 创建模型类
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
 
# 添加数据到数据库
user = User(name='John')
db.session.add(user)
db.session.commit()
 
# 查询数据
all_users = User.query.all()
 
# 更新数据
user = User.query.filter_by(name='John').first()
user.name = 'Jane'
db.session.commit()
 
# 删除数据
user = User.query.filter_by(name='Jane').first()
db.session.delete(user)
db.session.commit()

3、flask http 连接池

Flask 本身并不提供内置的 HTTP 连接池功能，但可以使用第三方库来实现在 Flask 中使用 HTTP 连接池。其中一个常用的库是 urllib3，它提供了高级的连接池管理功能。

示例：在 Flask 中使用 urllib3 来创建和管理 HTTP 连接池：

安装：pip install urllib3


from flask import Flask
import urllib3
 
app = Flask(__name__)
 
"""
使用 urllib3.PoolManager() 创建了一个连接池管理器对象 http，然后使用 http.request() 方法发送了一个 GET 请求。
您可以根据需要进行配置和自定义，例如设置最大连接数、超时时间、重试策略等。以下是一个示例，展示了如何进行自定义设置：
"""
 
 
@app.route('/index_1')
def index_1():
    http = urllib3.PoolManager()
    response = http.request('GET', 'http://api.example.com')
    return response.data
 
 
"""
对连接池进行了一些自定义配置，包括最大连接数、每个连接的最大数量、连接和读取的超时时间以及重试策略。
使用 urllib3 可以更好地控制和管理 HTTP 连接，提高 Flask 应用程序的性能和效率。
"""
 
 
@app.route('/index_2')
def index_2():
    http = urllib3.PoolManager(
        num_pools=10,  # 最大连接数
        maxsize=100,  # 每个连接的最大数量
        timeout=urllib3.Timeout(connect=2.0, read=5.0),  # 连接和读取的超时时间
        retries=urllib3.Retry(total=3, backoff_factor=0.1, status_forcelist=[500, 502, 503, 504])  # 重试策略
    )
    response = http.request('GET', 'http://api.example.com')
    return response.data

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/花生_TL007/article/detail/414252

Flask 数据库 连接池、DBUtils、http 连接池_flask mysql 连接池