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': 'nXMZ5JWbeKZ7N7RM',
    'host': '192.168.1.105',
    '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