环境 : python3.6 / win10 / vs2017 / sqlserver2017
一、需要安装的包pymssql
pip install pymssql
二、pymssql模块的介绍
pymssql 包 有modules:
-
pymssql
– 如果您关注DB-API遵从性,或者如果您习惯于DB-API语法,请使用它。 -
_mssql
– 比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 True
conn.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后,千万必须要关闭该连接。