lm_code/models/__init__.py

# from peewee import *
#
# # 连接到 SQLite 数据库，如果文件不存在会自动创建
# db = SqliteDatabase(r'E:\新建文件夹\lm_job\models\database.db')

import pymysql

from peewee import *
from playhouse.pool import PooledMySQLDatabase

pymysql.install_as_MySQLdb()

# 数据库配置
db_config = {
    'database': 'lm',
    'user': 'lm',
    'password': 'phzEDrRDEeNZ7yFX',
    'host': '192.168.1.79',
    'port': 3306
}

# 全局数据库实例
db1 = MySQLDatabase(
    db_config['database'],
    user=db_config['user'],
    password=db_config['password'],
    host=db_config['host'],
    port=db_config['port']
)

# class BaseModel(Model):
#     class Meta:
#         database = db1
#
#     def save(self, *args, **kwargs):
#         """在调用 save 时自动连接和关闭（若无事务）"""
#         db.connect(reuse_if_open=True)
#         try:
#             result = super().save(*args, **kwargs)
#         finally:
#             # 若当前没有事务且连接仍然打开，则关闭连接
#             if not db.in_transaction() and not db.is_closed():
#                 db.close()
#         return result
#
#     @classmethod
#     def get_or_create(cls, defaults=None, **kwargs):
#         """在调用 get_or_create 时自动连接和关闭（若无事务）"""
#         db.connect(reuse_if_open=True)
#         try:
#             obj, created = super().get_or_create(defaults=defaults, **kwargs)
#         finally:
#             # 若当前没有事务且连接仍然打开，则关闭连接
#             if not db.in_transaction() and not db.is_closed():
#                 db.close()
#         return obj, created