python SQLAlchemy 数据库连接池

前言

  如果您觉得有用的话，记得给博主点个赞，评论，收藏一键三连啊，写作不易啊^ _ ^。
  而且听说点赞的人每天的运气都不会太差，实在白嫖的话，那欢迎常来啊!!!

python SQLAlchemy 数据库连接池

SQLALchemy 链接数据库使用数据库连接池技术，原理是在系统初始化的时候，将数据库连接作为对象存储在内存中，当用户需要访问数据库时，并非建立一个新的链接，而是从链接池中取出一个已建立的空闲链接对象。使用完毕后，用户也并非将连接关闭，而是将连接放回连接池中，以供下一个请求访问使用。而链接的建立，断开都由链接池来管理，同时，还可以通过设置链接池的参数来控制链接池中的初始链接数，链接的上下限数以及每个链接的最大使用次数，最大空闲时间等。

1. 安装

安装SQLAlchemy

pip install SQLAlchemy

安装mysql

pip install pymysql

2. 创建数据库引擎

示例:

from sqlalchemy import create_engine

engine = create_engine(mysql_url, echo=True, pool_size=5, max_overflow=4, pool_recycle=7200, pool_timeout=30)

echo=True: 这表示在执行 SQL 查询时会输出所有 SQL 语句及其参数到控制台，方便调试。

pool_size=5: 这设置了数据库连接池的大小为 5，表示在连接池中最多可以保持 5 个连接。

max_overflow=4: 这允许在需要时，连接池外再创建最多 4 个额外的连接，超出连接池大小的部分会在使用后关闭。

pool_recycle=7200: 这表示连接在 7200 秒（2 小时）后会被回收，避免因长时间连接而导致的问题（例如，MySQL 的“互动超时”）。

pool_timeout=30: 这是连接池的超时时间，表示如果在 30 秒内没有获取到可用的连接，将会抛出异常。

3. 新建表，增删改查demo

配置文件:

sql.ini:

[datasource]
url = TESTDB:TESTDB@127.0.0.1:3306
db = TESTDB?charset=utf8

python demo:

from sqlalchemy import create_engine, Column, String, Integer, DateTime, Index, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import pathlib
import configparser

# 设置配置文件
current_dir = pathlib.Path(__file__).parent
config_file = current_dir / 'config' / 'sql.ini'
config = configparser.ConfigParser()
with open(config_file, 'r', encoding='utf-8') as f:
    config.read_file(f)

url = config['datasource']['url']
db = config['datasource']['db']

mysql_url = f'mysql+pymysql://{url}/{db}'
# 创建数据库引擎
"""
echo=True: 这表示在执行 SQL 查询时会输出所有 SQL 语句及其参数到控制台，方便调试。

pool_size=5: 这设置了数据库连接池的大小为 5，表示在连接池中最多可以保持 5 个连接。

max_overflow=4: 这允许在需要时，连接池外再创建最多 4 个额外的连接，超出连接池大小的部分会在使用后关闭。

pool_recycle=7200: 这表示连接在 7200 秒（2 小时）后会被回收，避免因长时间连接而导致的问题（例如，MySQL 的“互动超时”）。

pool_timeout=30: 这是连接池的超时时间，表示如果在 30 秒内没有获取到可用的连接，将会抛出异常。
"""
engine = create_engine(mysql_url, echo=True, pool_size=5, max_overflow=4, pool_recycle=7200, pool_timeout=30)

Base = declarative_base()

# 设置会话
Session = sessionmaker(bind=engine)
session = Session()


# 表结构
class YzyTest(Base):
    __tablename__ = 't_yzy_test'

    SEQUENCE_NO = Column(Integer, primary_key=True, autoincrement=True, comment='序列号')
    PK_STD_POINT_AI_RELATION = Column(String(36), unique=True, nullable=False, comment='id')
    FK_STD_AUDIT_POINT = Column(String(36), nullable=False, comment='审核标准id')
    FK_AI_STD = Column(String(36), nullable=False, comment='aiId')
    CHANNEL_TAG = Column(String(45), nullable=False, comment='渠道')
    FK_USER_CREATE = Column(String(36), nullable=True, comment='创建人id')
    USER_NAME_CREATE = Column(String(64), nullable=True, comment='创建人姓名')
    CREATE_TIME = Column(DateTime, default=text('CURRENT_TIMESTAMP'), nullable=False, comment='创建时间')

    __table_args__ = (
        Index('u_t_yzy_test_01', 'FK_STD_AUDIT_POINT', 'FK_AI_STD', 'CHANNEL_TAG', unique=True),
    )


# 创建表
def create_table():
    Base.metadata.create_all(engine)


# 查询数据
def query():
    return session.query(YzyTest).all()


# 插入数据
def save(param):
    session.add(param)
    session.commit()


# 更新数据
def update(param_id, updated_data):
    param = session.query(YzyTest).filter(YzyTest.PK_STD_POINT_AI_RELATION == param_id).first()
    if param:
        for key, value in updated_data.items():
            setattr(param, key, value)
        session.commit()


# 删除数据
def delete(param_id):
    param = session.query(YzyTest).filter(YzyTest.PK_STD_POINT_AI_RELATION == param_id).first()
    if param:
        session.delete(param)
        session.commit()


if __name__ == '__main__':
    create_table()

    # 示例用法：
    new_param = YzyTest(
        PK_STD_POINT_AI_RELATION='unique-id-1',
        FK_STD_AUDIT_POINT='audit-point-id',
        FK_AI_STD='ai-id',
        CHANNEL_TAG='channel-tag-example',
        USER_NAME_CREATE='创建者姓名'
    )
    save(new_param)

    params = query()
    for param in params:
        print(param.PK_STD_POINT_AI_RELATION, param.FK_STD_AUDIT_POINT, param.FK_AI_STD, param.CHANNEL_TAG, param.USER_NAME_CREATE)

    update('unique-id-1', {'CHANNEL_TAG': 'new_channel_tag'})
    params = query()
    for param in params:
        print(param.PK_STD_POINT_AI_RELATION, param.FK_STD_AUDIT_POINT, param.FK_AI_STD, param.CHANNEL_TAG, param.USER_NAME_CREATE)

    delete('unique-id-1')

测试: