网站首页 > 技术文章 正文
引言:
通过yaml文件配置数据库信息,实现多个数据库同时异步执行(增删改查)
在D盘新建一个后缀.yaml的文件 , 如文件Database.yaml内容:
Bash
-
host: 127.0.0.1
user: 用户名1
password: 密码1
db: 数据库名1
-
host: 127.0.0.2
user: 用户名2
password: 密码2
db: 数据库名2
Bash
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 类实现了以下功能:
- 多数据库连接支持:
该类支持连接多个 SQL Server 数据库。你可以在配置中提供多个数据库的连接信息,类会自动建立与这些数据库的连接。
- 自动执行 SQL 查询:
该类能够处理两种类型的 SQL 操作:查询操作 (SELECT):通过 fetch_query 方法执行查询操作并返回结果。非查询操作 (INSERT、UPDATE、DELETE 等):通过 execute_query 方法执行这些修改操作。
- 异步并行执行:
使用 Python 的 ThreadPoolExecutor,该类支持异步并行执行多个数据库的查询或更新操作。这意味着你可以同时向多个数据库发送查询请求,而无需等待每个请求单独完成。
- 自动区分 SQL 类型:
该类会自动判断 SQL 查询是查询操作(SELECT)还是更新操作(INSERT、UPDATE、DELETE等),并根据类型选择执行方式。通过 determine_query_type 方法自动识别查询类型。
- 错误处理:
类中有基本的错误处理机制,确保在连接数据库或执行 SQL 查询时如果发生错误,会输出相关的错误信息。
优点:这里读取yaml文件来获取数据库配置信息,使代码通用。当然也可以参考上一篇文章,用字典来获取数据库信息。
猜你喜欢
- 2024-12-15 使用 SQLAlchemy 实现 Python 数据库操作的 ORM 完整指南
- 2024-12-15 数据分析编程:SQL,Python or SPL?
- 2024-12-15 使用 Python 自动备份 SQL 数据库
- 2024-12-15 最全总结,聊聊 Python 数据处理全家桶(PgSQL篇)
- 06-24Python调用Docker API的使用方式(pycharm docker 调试)
- 06-24青少年Python编程系列28:Python中函数的递归调用
- 06-24python调用sqlite数据库案例(python 调用数据库)
- 06-24【Python机器学习系列】基于Flask来构建API调用机器学习模型服务
- 06-24通过pybind11来实现python调用C++接口(一)
- 06-24Python编程调用Deepseek API创建智能体
- 06-24python多装饰器针对函数、类、方法的调用顺序说明
- 06-24Python Qt GUI设计:Python调用UI文件的两种方法(基础篇—3)
- 271℃Python短文,Python中的嵌套条件语句(六)
- 269℃PythonNet:实现Python与.Net代码相互调用!
- 269℃python笔记:for循环嵌套。end=""的作用,图形打印
- 263℃Python操作Sqlserver数据库(多库同时异步执行:增删改查)
- 263℃Python实现字符串小写转大写并写入文件
- 122℃原来2025是完美的平方年,一起探索六种平方的算吧
- 103℃Python 和 JavaScript 终于联姻了!PythonMonkey 要火?
- 97℃Ollama v0.4.5-v0.4.7 更新集合:Ollama Python 库改进、新模型支持
- 最近发表
-
- Python调用Docker API的使用方式(pycharm docker 调试)
- 青少年Python编程系列28:Python中函数的递归调用
- python调用sqlite数据库案例(python 调用数据库)
- 【Python机器学习系列】基于Flask来构建API调用机器学习模型服务
- 通过pybind11来实现python调用C++接口(一)
- Python编程调用Deepseek API创建智能体
- python多装饰器针对函数、类、方法的调用顺序说明
- Python Qt GUI设计:Python调用UI文件的两种方法(基础篇—3)
- Python | Django 外部脚本调用 models 数据库
- 自学Python第九天——操作列表(python操作步骤)
- 标签列表
-
- python中类 (31)
- python 迭代 (34)
- python 小写 (35)
- python怎么输出 (33)
- python 日志 (35)
- python语音 (31)
- python 工程师 (34)
- python3 安装 (31)
- python音乐 (31)
- 安卓 python (32)
- python 小游戏 (32)
- python 安卓 (31)
- python聚类 (34)
- python向量 (31)
- python大全 (31)
- python次方 (33)
- python桌面 (32)
- python总结 (34)
- python浏览器 (32)
- python 请求 (32)
- python 前端 (32)
- python验证码 (33)
- python 题目 (32)
- python 文件写 (33)
- python中的用法 (32)