引言：

通过yaml文件配置数据库信息，实现多个数据库同时异步执行(增删改查)

在D盘新建一个后缀.yaml的文件 ， 如文件Database.yaml内容：

  -
    host: 127.0.0.1
    user: 用户名1
    password: 密码1
    db: 数据库名1
    
  -
    host: 127.0.0.2
    user: 用户名2
    password: 密码2
    db: 数据库名2

import os  # 导入 os 模块，用于操作系统相关的功能（在当前代码中未使用，可以考虑去除）
import pymssql  # 导入 pymssql 模块，用于连接 SQL Server 数据库
import yaml
from concurrent.futures import ThreadPoolExecutor, as_completed  # 导入并发执行相关模块，支持线程池管理和异步执行


class DBConnector:
    """"说明： 此类实现直接执行所有数据库的增删该查，不区分，配置多少库就执行多少库    """""
    def __init__(self):
        self.config = self.load_config()  # 如果没有传入配置，则使用默认配置，这里配置了多个数据库
        self.connections = []  # 用来存储多个数据库连接
        self.setup_connections()  # 初始化所有数据库连接

    def load_config(self):
        config_file = r'D:\Database.yaml'     #yaml文件路径
        with open(config_file, 'r') as file:
            return yaml.safe_load(file)

    def setup_connections(self):
        """初始化所有数据库连接"""
        # 遍历配置的所有数据库连接配置，连接每个数据库
        for db_config in self.config:
            try:
                # 使用 pymssql.connect 创建数据库连接
                connection = pymssql.connect(
                    server=db_config['host'],  # 数据库主机地址
                    user=db_config['user'],  # 用户名
                    password=db_config['password'],  # 密码
                    database=db_config['db'],  # 数据库名
                    charset='cp936',  # 字符集(根据实际库的填写)
                    as_dict=True  # 返回的查询结果将是字典格式
                )
                self.connections.append(connection)  # 将连接添加到 connections 列表中
                # print(f"成功连接数据库: {db_config['serverid']}")  # 打印连接成功的信息
            except Exception as e:
                print(f"连接数据库失败: {db_config['serverid']} - {e}")  # 捕获异常并打印错误信息

    def execute_query(self, connection, query, params=None):
        """执行更新、删除、插入等非SELECT查询"""
        try:
            with connection.cursor() as cursor:  # 获取数据库连接的游标对象
                cursor.execute(query, params)  # 执行传入的 SQL 查询
                connection.commit()  # 提交事务，确保数据更新被保存
        except Exception as e:
            print(f"执行查询失败: {e}")  # 捕获异常并打印错误信息
            raise e  # 将异常重新抛出，确保调用者能处理

    def fetch_query(self, connection, query, params=None):
        """执行SELECT查询并返回结果"""
        try:
            with connection.cursor() as cursor:  # 获取数据库连接的游标对象
                cursor.execute(query, params)  # 执行传入的 SQL 查询
                results = cursor.fetchall()  # 获取查询的所有结果
                return results  # 返回查询结果
        except Exception as e:
            print(f"执行查询失败: {e}")  # 捕获异常并打印错误信息
            raise e  # 将异常重新抛出，确保调用者能处理

    def determine_query_type(self, query):
        """自动判断查询类型"""
        query_upper = query.strip().upper()  # 将查询转换为大写，去除多余的空格
        if query_upper.startswith("SELECT"):  # 如果是 SELECT 查询
            return 'fetch'  # 返回 'fetch' 表示是查询操作
        else:
            return 'execute'  # 否则返回 'execute' 表示是修改操作（INSERT, UPDATE, DELETE等）

    def async_execute(self, query, params=None):
        """异步执行SQL查询，支持多库操作"""
        query_type = self.determine_query_type(query)  # 判断查询类型（是查询还是修改）
        results = []  # 用于存储所有数据库执行后的结果
        futures = []  # 用于存储所有数据库任务的 futures 对象

        # 使用 ThreadPoolExecutor 来并行执行数据库操作
        with ThreadPoolExecutor() as executor:
            for connection in self.connections:  # 遍历所有数据库连接
                if query_type == 'fetch':  # 如果是查询操作
                    # 提交异步任务执行查询，返回的结果是查询结果
                    futures.append(executor.submit(self.fetch_query, connection, query, params))
                else:  # 如果是更新、插入等修改操作
                    # 提交异步任务执行修改操作
                    futures.append(executor.submit(self.execute_query, connection, query, params))

            # 等待所有任务完成，并收集结果
            for future in as_completed(futures):  # 遍历所有已完成的任务
                try:
                    result = future.result()  # 获取任务的执行结果
                    if result is not None:
                        results.append(result)  # 将查询结果存储到 results 列表中
                except Exception as e:
                    print(f"执行任务失败: {e}")  # 捕获任务执行中的异常并打印

        return results  # 返回所有数据库的查询结果

if __name__ == "__main__":
    pass
    # 示例：连接到多个数据库并执行查询
    # connector = DBConnector()  # 初始化数据库连接器，自动连接所有数据库
    # # 执行更新操作（例如插入、更新、删除等）
    # update_query = 'UPDATE sysconfig SET lastsysdate = "20241205"'  # 需要执行的更新 SQL
    # connector.async_execute(update_query)  # 异步执行更新操作
    #执行查询操作
    # select_query = 'SELECT * FROM sysconfig'  # 需要执行的查询 SQL
    # select_results = connector.async_execute(select_query)  # 异步执行查询操作
    # #打印查询结果
    # for result in select_results:
    #     print(result)

DBConnector 类实现了以下功能：

1. 多数据库连接支持：

该类支持连接多个 SQL Server 数据库。你可以在配置中提供多个数据库的连接信息，类会自动建立与这些数据库的连接。

2. 自动执行 SQL 查询：

该类能够处理两种类型的 SQL 操作：查询操作 (SELECT)：通过 fetch_query 方法执行查询操作并返回结果。非查询操作 (INSERT、UPDATE、DELETE 等)：通过 execute_query 方法执行这些修改操作。

3. 异步并行执行：

使用 Python 的 ThreadPoolExecutor，该类支持异步并行执行多个数据库的查询或更新操作。这意味着你可以同时向多个数据库发送查询请求，而无需等待每个请求单独完成。

4. 自动区分 SQL 类型：

该类会自动判断 SQL 查询是查询操作（SELECT）还是更新操作（INSERT、UPDATE、DELETE等），并根据类型选择执行方式。通过 determine_query_type 方法自动识别查询类型。

5. 错误处理：

类中有基本的错误处理机制，确保在连接数据库或执行 SQL 查询时如果发生错误，会输出相关的错误信息。

优点：这里读取yaml文件来获取数据库配置信息，使代码通用。当然也可以参考上一篇文章，用字典来获取数据库信息。