环境 : python3.6 / win10 / vs2017 / sqlserver2017
一、需要安装的包pymssql
pip install pymssql
二、pymssql模块的介绍
pymssql 包 有modules:
- – 如果您关注DB-API遵从性，或者如果您习惯于DB-API语法，请使用它。
- – 比pymssql更高性能和易用性，性能高出不是一点点，用法也相对简单。
所以我更加推荐使用_mssql，而不是网上案列里比较多的pymssql
三、对_mssql模块的封装
1、简单的执行
class C_SQLServer(object): def __init__(self, Server,user,password,database): self.Server=Server self.user=user self.password=password self.database=database #执行无返回操作，适用与(insert,update,delete) def execute_non_query(self,SQLStr): conn = _mssql.connect(server=self.Server, user=self.user, password=self.password, database=self.database,charset='utf8') conn.execute_non_query(SQLStr) #执行返回迭代器的操作，迭代器中的行以字典方式展示，适用于(select) def execute_query(self,SQLStr): conn = _mssql.connect(server=self.Server, user=self.user, password=self.password, database=self.database,charset='utf8') conn.execute_query(SQLStr) return conn #执行返回单行的字典 def execute_row(self,SQLStr): conn = _mssql.connect(server=self.Server, user=self.user, password=self.password, database=self.database,charset='utf8') row=conn.execute_row(SQLStr) return row #执行返回单值的操作，适用于返回行计数等 def execute_scalar(self,SQLStr): conn = _mssql.connect(server=self.Server, user=self.user, password=self.password, database=self.database,charset='utf8') scalar=conn.execute_scalar(SQLStr) return scalar #获取标题，以及标题类型字典 def MSSQL_GetTitleDict(self,conn): titleDict={} for rows in conn.get_header(): titleDict[rows[0]]=rows[1] #如果调用conn完成后千万记得，要吧连接关闭。 conn.close() return titleDict 四、对于大批量Insert的操作
pymssql的模块提供了executemany来执行大批量的导入。
导入列表里的元素行为Tuple，类似 DataList=[(1,2),(2,3)]
cursor.executemany( "INSERT INTO persons VALUES (%d, %s, %s)", [(1, 'John Smith', 'John Doe'), (2, 'Jane Doe', 'Joe Dog'), (3, 'Mike T.', 'Sarah H.')])# you must call commit() to persist your data if you don't set autocommit to Trueconn.commit()
_mssql模块没有提供批量导入的功能。
但是我们可以用拼接字符串 Insert ————Select————UNION ALL————SELECT 去实现。
测经过测试，同样插入10W的数据，_mssql模块写拼接比pymssql的executemany快了近3倍多。
代码如下：
def GetTableTitle(self,tableName): SQLStr=f"select * from {tableName}" conn=self.execute_query(f"select * from {tableName}") titleDict=self.MSSQL_GetTitleDict(conn) return titleDict #拼接字符串 Insert ————Select————UNION ALL————SELECT类型插入 def InsertByRow(self,tableName,TitleList,DataList): #获取列头的字典包含列名以及数据类型 titleDict=self.GetTableTitle(tableName) #定义Insert语句的头部 insertTitleStr=f"Insert into {tableName} (" + ','.join(TitleList)+")\n" #批导入变量，执行行号 i=0 #按行执行 for row_dict in DataList: insertRowStrList=[] #循环列 for columnName in TitleList: columnType=titleDict[columnName] if columnType in [1,4]: isChar=1 else: isChar=0 columValue= row_dict[columnName] #SqlParameter_AddQuotes函数用以增给值增加单引号 columValue=SqlParameter_AddQuotes(isChar,columValue) queryStr=columnName+"="+columValue insertRowStrList.append(queryStr) #行的SelectStr insertRowStr=','.join(insertRowStrList) if i==0: insertStr="Select "+insertRowStr else: insertStr+="\n union all \nSelect "+insertRowStr i+=1 #定义批量插入的大小，这里是300行为一批Insert if i%300==0: self.execute_non_query(insertTitleStr+'\n'+insertStr) i=0 #剩余数据Insert self.execute_non_query(insertTitleStr+'\n'+insertStr)
性能刚刚的！日常的操作基本也就都封装好了！
但是记得打开conn后，千万必须要关闭该连接。