基于Python+Peewee实现多MySQL数据库ORM交互的技术问询

嘿，我来帮你搞定这个多数据库交互的问题！结合你用的Peewee 2.10和Python 2.7，我给你梳理一套完整的实现方案，完全覆盖你需求的所有步骤：

首先要为MainDb和类型数据库分别创建数据库连接实例，再绑定对应的模型。因为你的类型数据库是根据插入的type字段动态确定的，我们可以先定义MainDb的固定模型，再通过函数动态生成类型数据库的连接和模型。

先导入必要模块并配置MainDb：

from peewee import MySQLDatabase, Model, CharField, IntegerField
import MySQLdb  # 匹配你环境里的MySQLdb驱动

定义MainDb的连接与types表模型：

# 配置MainDb连接信息，替换成你的实际参数
main_db = MySQLDatabase(
    'main_db_name',
    user='your_username',
    password='your_password',
    host='localhost',
    port=3306,
    thread_safe=True,
    db_driver=MySQLdb
)

# MainDb的types表模型
class MainType(Model):
    name = CharField()
    type = CharField()  # 用来确定后续要连接的类型数据库

    class Meta:
        database = main_db
        table_name = 'types'

写一个函数，根据type字段动态生成对应数据库的连接和模型（你可以根据实际的数据库/表命名规则调整）：

def get_type_db_and_model(type_name):
    # 假设类型数据库命名为 type_{type_name}_db
    type_db_name = f"type_{type_name}_db"
    type_db = MySQLDatabase(
        type_db_name,
        user='your_username',
        password='your_password',
        host='localhost',
        port=3306,
        thread_safe=True,
        db_driver=MySQLdb
    )

    # 假设对应类型的表命名为 type_{type_name}_data
    class TypeData(Model):
        main_id = IntegerField()  # 关联MainDb的自增ID
        data = CharField()  # 替换成你实际需要插入的字段

        class Meta:
            database = type_db
            table_name = f"type_{type_name}_data"
    
    return type_db, TypeData

用try...finally块确保无论操作成功还是抛出异常，都会删除两边数据库的对应条目，同时用Peewee的原子操作保证单库内的操作一致性：

def process_data(name, type_name, type_data_content):
    main_id = None
    type_db = None
    TypeData = None

    try:
        # 1. 插入MainDb并获取自增ID
        main_db.connect()
        with main_db.atomic():
            # Peewee会自动填充自增ID，直接访问实例的id属性即可
            main_type = MainType.create(name=name, type=type_name)
            main_id = main_type.id
            print(f"获取到MainID: {main_id}")

        # 2. 连接对应类型数据库，插入关联数据
        type_db, TypeData = get_type_db_and_model(type_name)
        type_db.connect()
        with type_db.atomic():
            TypeData.create(main_id=main_id, data=type_data_content)
            print(f"已在类型数据库 {type_db.database} 中插入关联数据")

        # 3. 执行带MainID的查询操作（示例）
        if main_id:
            query_result = TypeData.select().where(TypeData.main_id == main_id).get()
            print(f"查询结果：main_id={query_result.main_id}，data={query_result.data}")

        # 这里可以添加你需要的其他业务逻辑

    except Exception as e:
        print(f"操作过程中出现异常: {str(e)}")
        # 可以选择在这里处理异常，或者向上抛出
        raise
    finally:
        # 4. 无论成功失败，强制清理两边的数据
        print("开始清理数据...")
        # 清理类型数据库的条目
        if type_db and TypeData and main_id:
            try:
                with type_db.atomic():
                    delete_count = TypeData.delete().where(TypeData.main_id == main_id).execute()
                    print(f"类型数据库中删除了 {delete_count} 条记录")
            except Exception as e:
                print(f"删除类型数据库条目失败: {str(e)}")
            finally:
                if not type_db.is_closed():
                    type_db.close()

        # 清理MainDb的条目
        if main_id:
            try:
                with main_db.atomic():
                    delete_count = MainType.delete().where(MainType.id == main_id).execute()
                    print(f"MainDb中删除了 {delete_count} 条记录")
            except Exception as e:
                print(f"删除MainDb条目失败: {str(e)}")
            finally:
                if not main_db.is_closed():
                    main_db.close()

# 调用示例
if __name__ == '__main__':
    process_data("测试名称", "user", "测试用户数据内容")

• 自增ID的获取：Peewee在调用Model.create()后，返回的实例会自动填充自增主键字段（默认是id），完全不需要像原生MySQLdb那样手动获取cursor.lastrowid，更简洁可靠。
• 多数据库连接管理：每个数据库连接要单独处理连接和关闭，finally块里的关闭操作能避免连接泄漏；atomic()方法确保单库内的操作要么全部成功，要么全部回滚。
• 异常处理逻辑：try...finally的结构保证了不管中间出现任何异常（比如插入失败、查询报错），都会执行清理步骤。如果需要更严格的跨库事务一致性（比如两个库的操作要么都成功要么都失败），可以研究MySQL的XA分布式事务，但配置相对复杂，你的场景用finally做最终清理已经足够。
• 动态模型的灵活性：通过函数动态生成类型数据库的连接和模型，能适配不同type字段对应的数据库，你可以根据实际的命名规则调整函数内的逻辑。

内容的提问来源于stack exchange，提问作者user7014262