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ù)庫引擎
"""
echo=True: 這表示在執(zhí)行 SQL 查詢時會輸出所有 SQL 語句及其參數(shù)到控制臺，方便調(diào)試。

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

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

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

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

Base = declarative_base()

# 設(shè)置會話
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='序列號')
    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='創(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)建時間')

    __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')