1. AI技术聚合首页
  2. Python

Python与SQL Server数据库完美结合，数据处理得心应手

心中带点小风骚 • • Python • 阅读 30

方案一：
直接套用脚本，需可以看懂一些脚本逻辑

这个类实现了同时连接多个 SQL Server 数据库，并提供了执行查询和非查询操作的能力

import pymssql

class SqlServerConnector:
    def __init__(self, servers):
        self.connections = {}  # 存储数据库连接的字典
        for server in servers:
            conn = pymssql.connect(
                server=server['host'],
                port=server['port'],
                user=server['username'],
                password=server['password'],
                database=server['database']
            )
            self.connections[server['name']] = conn

    def execute_query(self, connection_name, query, params=None):
        with self.connections[connection_name].cursor() as cursor:  # 获取游标(cursor)
            cursor.execute(query, params)  # 执行查询语句
            result = cursor.fetchall()  # 获取结果集
        return result
    
    def execute_non_query(self, connection_name, query, params=None):
        with self.connections[connection_name].cursor() as cursor:  # 获取游标(cursor)
            cursor.execute(query, params)  # 执行增删改语句
            self.connections[connection_name].commit()  # 提交事务
            return cursor.rowcount  # 返回受影响的行数
        
    def execute_query_all(self, servers, query, params=None):
        results = {}  # 存储所有查询结果的字典
        for server in servers:
            connection_name = server['name']
            results[connection_name] = self.execute_query(connection_name, query, params)
        return results
    
    def execute_non_query_all(self, servers, query, params=None):
        results = {}  # 存储所有操作结果的字典
        for server in servers:
            connection_name = server['name']
            try:
                res = self.execute_non_query(connection_name, query, params)
                results[connection_name] = res
            except Exception as e:
                print(f"Error occurred when running query on {connection_name}: {e}")
                self.connections[connection_name].rollback() # 出错时回滚
        return results

以下是代码中的解释

import pymssql：导入 pymssql 库，用于与 SQL Server 数据库进行连接和交互。

class SqlServerConnector:：定义一个名为 SqlServerConnector 的类。

def __init__(self, servers):：类的初始化方法，接受一个包含多个数据库信息的列表作为参数。

self.connections = {}：初始化一个字典，用于存储数据库连接。

for server in servers:：遍历数据库列表中的每个数据库信息。

conn = pymssql.connect(...)：根据数据库信息创建数据库连接对象。server['host']：数据库服务器的主机地址。server['port']：数据库服务器的端口号。server['username']：连接数据库的用户名。server['password']：连接数据库的密码。server['database']：要连接的数据库名称。

self.connections[server['name']] = conn：将数据库连接对象添加到字典中，以数据库名称作为键。

def execute_query(self, connection_name, query, params=None):：执行查询语句的方法。connection_name：数据库名称。query：要执行的查询语句。params=None：可选的查询参数。
def __init__(self, servers):：类的初始化方法，接受一个包含多个数据库信息的列表作为参数。

self.connections = {}：初始化一个字典，用于存储数据库连接。

for server in servers:：遍历数据库列表中的每个数据库信息。

conn = pymssql.connect(...)：根据数据库信息创建数据库连接对象。server['host']：数据库服务器的主机地址。server['port']：数据库服务器的端口号。server['username']：连接数据库的用户名。server['password']：连接数据库的密码。server['database']：要连接的数据库名称。

self.connections[server['name']] = conn：将数据库连接对象添加到字典中，以数据库名称作为键。

def execute_query(self, connection_name, query, params=None):：执行查询语句的方法。connection_name：数据库名称。query：要执行的查询语句。params=None：可选的查询参数。

方案二：直接调用封装脚本（写用例，执行脚本即可）
脚本实现封装后，只需要在SqlServer.yaml文件中写用例即可，此后执行SqlServer.py脚本即实现数据库的增删改查操作

目录介绍：

SqlServer.yaml 编写多个数据库连接信息

Sqlserver_jzjy_uat:
  -
    serverid: 0
    host: 127.0.0.1
    user: sa
    password: pwd1
    db: run
    charset: utf-8

  -
    serverid: 1
    host: 127.0.0.2
    user: sa
    password: pwd2
    db: run
    charset: utf-8

  -
    serverid: 2
    host: 127.0.0.3
    user: sa
    password: pwd2
    db: run
    charset: utf-8

PublicConfig.py脚本： 配置读取信息，方便调用

import os
from Public_Utils.util_yaml import YamlReader

class YamlPath:
    def __init__(self):
        current = os.path.abspath(__file__)
        self.base_dir = os.path.dirname(os.path.dirname(current))
        self._config_path = self.base_dir + os.sep + "Public_Config\Public_yaml"

    def get_sqlserver_file(self):
        self._config_file = self._config_path + os.sep + "SqlServer.yaml"
        return self._config_file

class ConfigYaml:
    def __init__(self):   #初始yaml读取配置文件
        self.sqlserver_config = YamlReader(YamlPath().get_sqlserver_file()).yaml_data()

    def get_sqlserver_yaml(self):
        return self.sqlserver_config['Sqlserver_jzjy_uat']

    def get_serveridlist_yaml(self):
        return self.sqlserver_config['uat_serverlist']


if __name__ == '__main__':
    pass


SqlServer.py 执行脚本

import pymssql
from Public_Config.PublicConfig import ConfigYaml

class SQLServer:
    def __init__(self):
        # 连接数据库
        self.connections = {}
        for server in ConfigYaml().get_sqlserver_yaml():
                conn = pymssql.connect(host =server['host'],
                                       user=server['user'],
                                       password=server['password'],
                                       database=server['db'],
                                       charset='cp936', as_dict=True)
                self.connections[server['serverid']] = conn

    def execute_query(self, connection_name, query, params=None):
        with self.connections[connection_name].cursor() as cursor:     #获取游标（cursor）
            cursor.execute(query,params)   #执行查询语句
            result = cursor.fetchall()     #获取结果集合
        return result

    def execute_query_all(self, query, params=None):  #全部数据库执行查询（select）
        results = {}
        for serverid in ConfigYaml().get_sqlserver_yaml():
            connection_name = serverid['serverid']
            results[connection_name] = self.execute_query(connection_name,query,params)
        return results

    def execute_query_special(self, query, params=None):  #指定的几个数据库一起执行语句 （select）
        results = {}
        serveridlist = ConfigYaml().get_serveridlist_yaml()['serveridlist']
        for serverid in ConfigYaml().get_sqlserver_yaml():
            if str(serverid['serverid']) in serveridlist.split(','):
                connection_name = serverid['serverid']
                results[connection_name] = self.execute_query(connection_name, query, params)
        return results

    def execute_non_query(self,connection_name,query,params=None):
        with self.connections[connection_name].cursor() as cursor:
            cursor.execute(query,params)    #执行增删查等操作
            self.connections[connection_name].commit()   #提交事务
        return cursor.rowcount      #返回受影响的行数

    def execute_non_query_all(self, query, params=None):  #实现配置表中的所有增删改等操作insert,delete,update）
        results = {}   #储存所有查询结果的字典
        for serverid in ConfigYaml().get_sqlserver_yaml():
            connection_name = serverid['serverid']
            try:
                res = self.execute_non_query(connection_name,query,params)
                results[connection_name] = res
            except Exception as e:
                print(f"Error occurred when running query on {connection_name} :{e}")
                self.connections[connection_name].rollback()   #出错时回滚
        return results

    def execute_non_query_special(self, query, params=None):  #实现配置表中指定数据库的增删改等操作insert,delete,update）
        results = {}   #储存所有查询结果的字典
        for serverid in ConfigYaml().get_sqlserver_yaml():
            serveridlist = ConfigYaml().get_serveridlist_yaml()['serveridlist']
            if str(serverid['serverid']) in serveridlist.split(','):
                connection_name = serverid['serverid']
                try:
                    res = self.execute_non_query(connection_name,query,params)
                    results[connection_name] = res
                except Exception as e:
                    print(f"Error occurred when running query on {connection_name} :{e}")
                    self.connections[connection_name].rollback()   #出错时回滚
        return results

if __name__ == '__main__':
    pass

util_yaml.py

import os
import yaml

class YamlReader:
    #初始化，判断文件是否存在
    def __init__(self,yaml_file):
        if os.path.exists(yaml_file):
            self.yaml_file = yaml_file
        else:
            raise FileNotFoundError("yaml文件不存在")
        self._data = None
        self._data_all = None

    def yaml_data(self):  #yaml文件读取 --单个文档读取
        #第一次调用data,读取yaml文档，如果不是，直接返回之前保存的数据
        if not self._data:
            with open(self.yaml_file,'rb') as f:
                self._data = yaml.safe_load(f)
        return self._data

    def yaml_data_all(self):  #多个文档的读取
        if not self._data_all:
            with open(self.yaml_file,'rb') as f:
                self._data_all = yaml.safe_load_all(f)
        return self._data_all

文章出处登录后可见！

已经登录？立即刷新

共计人评分，平均分

到目前为止还没有投票！成为第一位评论此文章。

赞 (0)
心中带点小风骚的头像心中带点小风骚普通用户
0
上一篇 2023年12月28日
下一篇 2023年12月28日

相关推荐