【一】引入

【1】正常创建数据库对象

import uuid
import pymysql
from flask import Flask, g, request, session
from flask_session import RedisSessionInterface
from pymysql.cursors import DictCursor
from redis import Redis
from flask_session import Session

app = Flask(__name__)
app.debug = True
app.secret_key = uuid.uuid4()

# （1）创建链接对象
conn = pymysql.connect(
    user="root",
    password="1314521",
    host="localhost",
    port=3306,
    database="test"
)

# （2）创建 cursor 对象
# DictCursor : 序列化查到的数据是字典格式
cursor = conn.cursor(cursor=DictCursor)


@app.route('/', methods=['GET', 'POST'])
def index():
    cursor.execute("SELECT * FROM test WHERE ID < 10")
    res = cursor.fetchall()
    print(res)
    return 'index'


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

'''
问题：
   conn 和 cursor 是全局的，多个进程同时操作 数据库会发生数据库错乱
   conn 和 cursor 要在视图函数中独立生成
   
   # 在Django中使用orm语句时，会创建一个链接对象，执行，执行完毕后释放链接对象
'''

• 会产生问题
  • conn 和 cursor 是全局的，多个进程同时操作 数据库会发生数据库错乱
  • conn 和 cursor 要在视图函数中独立生成

在Django中使用orm语句时，会创建一个链接对象，执行，执行完毕后释放链接对象

• 优化后

import uuid
import pymysql
from flask import Flask, g, request, session
from flask_session import RedisSessionInterface
from pymysql.cursors import DictCursor
from redis import Redis
from flask_session import Session

app = Flask(__name__)
app.debug = True
app.secret_key = uuid.uuid4()

@app.route('/', methods=['GET', 'POST'])
def index():
    # （1）创建链接对象
    conn = pymysql.connect(
        user="root",
        password="1314521",
        host="localhost",
        port=3306,
        database="test"
    )

    # （2）创建 cursor 对象
    # DictCursor : 序列化查到的数据是字典格式
    cursor = conn.cursor(cursor=DictCursor)
    cursor.execute("SELECT * FROM test WHERE ID < 10")
    res = cursor.fetchall()
    print(res)
    cursor.close()
    return 'index'


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

    '''
问题：
   conn 和 cursor 是全局的，多个进程同时操作 数据库会发生数据库错乱
   conn 和 cursor 要在视图函数中独立生成

   # 在Django中使用orm语句时，会创建一个链接对象，执行，执行完毕后释放链接对象
'''

【二】数据库连接池

【1】安装

pip install dbutils

【2】封装数据库连接池单例

• setting.py

# -*-coding: Utf-8 -*-
# @File : 06pool .py
# author: Chimengmeng
# blog_url : https://www.cnblogs.com/dream-ze/
# Time：2023/8/24
import pymysql
from dbutils.pooled_db import PooledDB, SharedDBConnection
from redis import Redis


class Config(object):
    DEBUG = True
    SECRET_KEY = "umsuldfsdflskjdf"
    PERMANENT_SESSION_LIFETIME = timedelta(minutes=20)
    SESSION_REFRESH_EACH_REQUEST = True
    SESSION_TYPE = "redis"

    # 实例化得到 PYMYSQL_POOL  对象
    # 池 为 6 的 数据库连接池
    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',
        port=3306,
        user='root',
        password='123456',
        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,sql, args,cursor =pymysql.cursors.DictCursor):
        conn, cursor = cls.open(cursor)
        cursor.execute(sql, args)
        obj = cursor.fetchall()
        cls.close(conn, cursor)
        return obj

• 使用

obj = SQLHelper.fetch_one("select id,name from users where name=%(user)s and pwd=%(pwd)s", form.data)

【补充】Django中使用数据库连接池

• 在Django中使用数据库连接池可以提高应用程序的性能和可扩展性。
• 数据库连接池是管理数据库连接的工具，它可以重复使用数据库连接，减少每次请求时创建和销毁连接的开销。
• 下面是使用数据库连接池的步骤：

【1】安装数据库连接池库

• 首先，确保你已经安装了数据库驱动程序，比如psycopg2（用于PostgreSQL）或mysqlclient（用于MySQL）。

• 然后，你可以选择一个合适的数据库连接池库，比如django-db-connection-pool。

• 你可以通过pip包管理器来安装所选的库：

pip install django-db-connection-pool

【2】配置数据库连接池

• 在Django项目的设置文件（settings.py）中进行相应的配置。
• 一般来说，你需要添加数据库连接池相关的设置项，比如最小连接数、最大连接数等。
• 以下是一个示例配置，你可以根据自己的需求进行调整：

# settings.py

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'your_database_name',
        'USER': 'your_username',
        'PASSWORD': 'your_password',
        'HOST': 'your_host',
        'PORT': 'your_port',

        # 添加数据库连接池相关设置项
        'CONN_MAX_AGE': 60,  # 指定连接在空闲多长时间后被关闭，单位为秒
        'POOL_SIZE': 10,  # 连接池中连接的最大数量
        'MIN_POOL_SIZE': 2  # 连接池中连接的最小数量，用于处理低负载场景
    }
}

• 以上配置示例中，我们使用了PostgreSQL作为数据库引擎，你可以根据自己的需要选择合适的数据库引擎。
• CONN_MAX_AGE设置了连接在空闲60秒后被关闭，以避免长时间空闲的连接占用资源。
• POOL_SIZE表示连接池中连接的最大数量，可以根据应用程序的负载情况进行调整。
• MIN_POOL_SIZE设置连接池中连接的最小数量，用于处理低负载场景。

【3】使用数据库连接池

• 一旦配置好数据库连接池，你可以像平常一样使用Django的数据库API来访问数据库。
  • 连接池会在需要的时候自动创建和重复使用数据库连接，无需手动管理。
• 以下是一个简单的示例，假设你有一个Book模型类：

from django.db import models

class Book(models.Model):
    title = models.CharField(max_length=100)
    author = models.CharField(max_length=100)

# 进行数据库操作
books = Book.objects.all()
for book in books:
    print(book.title)

• 在上述示例中，我们使用objects.all()来获取所有的书籍，并遍历打印每本书的标题。
  • 连接池会在需要时自动分配数据库连接，以执行查询操作。
• 通过使用数据库连接池，你可以提高数据库访问的效率和可扩展性，特别是在高流量的应用程序中。
  • 连接池会智能管理连接的创建和销毁，帮助减少了数据库连接的开销，提高了应用程序的性能。

本栏目推荐文章
• Jmeter 如何连接mysql数据库？
• SQL Server 检测是不是数字型数据的两种方法
• 字节微服务HTTP框架Hertz使用与源码分析｜拥抱开
• 如何通过Navicat把MySQL远程数据库导入到本地数据库
• 和利时MACS6通过dat文件实现任意时间范围与测点的历史数据提取与科研分析
• 深入理解spring框架：剖析多线程模式下数据库连接
• 数据可视化是如何帮助大型企业提高效率的？
• redis 数据库
• openpyxl模块--------------------------------重复数据
• Django客户端应用1向服务端应用2发送POST请求并接收解析数据

框架 数据库 数据 Flask 10.0flask_migrate框架migrate数据库 框架 数据库 数据flask flask_marshmallow序列marshmallow数据 wtforms方式 数据flask 数据库 数据flask sqlalchemy数据表flask-script信号 数据流 数据flask 数据python flask json sqlalchemy项目 数据表flask 数据表flask flask-script sqlalchemy