pip install SQLAlchemy

pip install pymysql

from sqlalchemy import create_engine

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

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

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

# 設(shè)置配置文件
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}'
# 創(chuàng)建數(shù)據(jù)庫(kù)引擎
"""
echo=True: 這表示在執(zhí)行 SQL 查詢時(shí)會(huì)輸出所有 SQL 語(yǔ)句及其參數(shù)到控制臺(tái)，方便調(diào)試。

pool_size=5: 這設(shè)置了數(shù)據(jù)庫(kù)連接池的大小為 5，表示在連接池中最多可以保持 5 個(gè)連接。

max_overflow=4: 這允許在需要時(shí)，連接池外再創(chuàng)建最多 4 個(gè)額外的連接，超出連接池大小的部分會(huì)在使用后關(guān)閉。

pool_recycle=7200: 這表示連接在 7200 秒（2 小時(shí)）后會(huì)被回收，避免因長(zhǎng)時(shí)間連接而導(dǎo)致的問(wèn)題（例如，MySQL 的“互動(dòng)超時(shí)”）。

pool_timeout=30: 這是連接池的超時(shí)時(shí)間，表示如果在 30 秒內(nèi)沒(méi)有獲取到可用的連接，將會(huì)拋出異常。
"""
engine = create_engine(mysql_url, echo=True, pool_size=5, max_overflow=4, pool_recycle=7200, pool_timeout=30)

Base = declarative_base()

# 設(shè)置會(huì)話
Session = sessionmaker(bind=engine)
session = Session()


# 表結(jié)構(gòu)
class YzyTest(Base):
    __tablename__ = 't_yzy_test'

    SEQUENCE_NO = Column(Integer, primary_key=True, autoincrement=True, comment='序列號(hào)')
    PK_STD_POINT_AI_RELATION = Column(String(36), unique=True, nullable=False, comment='id')
    FK_STD_AUDIT_POINT = Column(String(36), nullable=False, comment='審核標(biāo)準(zhǔn)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='創(chuàng)建人id')
    USER_NAME_CREATE = Column(String(64), nullable=True, comment='創(chuàng)建人姓名')
    CREATE_TIME = Column(DateTime, default=text('CURRENT_TIMESTAMP'), nullable=False, comment='創(chuàng)建時(shí)間')

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


# 創(chuàng)建表
def create_table():
    Base.metadata.create_all(engine)


# 查詢數(shù)據(jù)
def query():
    return session.query(YzyTest).all()


# 插入數(shù)據(jù)
def save(param):
    session.add(param)
    session.commit()


# 更新數(shù)據(jù)
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()


# 刪除數(shù)據(jù)
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='創(chuàng)建者姓名'
    )
    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')