Python操作MSSQL
python连接SQL Server数据库 - pymssql使用基础：https://www.cnblogs.com/baiyangcao/p/pymssql_basic.html
廖雪峰官网 之 Python 访问数据库（SQLLite / Mysql / SQLAlchemy）
python-整理--连接MSSQL：https://www.cnblogs.com/mirrortom/p/5218865.html
Python 操作SQLServer 需要使用 pymssql 模块，使用pip install pymssql安装。然后import该包即可。安装成功后，使用如下语句和 MSSql 数据库交互。
（ pymssql 需要安装 Cython：https://pypi.org/project/Cython 和 freetds：linux下利用freetds 访问sqlserver数据库 ）
下载 FreeTDS，地址：www.freetds.org
下载后解压: tar -zxvf XXX.tar.gz
然后执行
./configure --prefix=/usr/local/freetds --with-tdsver=7.1 --enable-msdblib
make
make install
示例代码（游标使用注意事项）：
import pymssql
# 可以进入 connect 方法里面查看更多参数
conn=pymssql.connect(server='192.168.0.110', port=9526, user='sa', passWord='pwd', database='test')
# 游标使用注意事项
# 一个连接一次只能有一个游标的查询处于活跃状态，如下：
cursor_1 = conn.cursor()
cursor_1.execute('SELECT * FROM persons')
cursor_2 = conn.cursor()
cursor_2.execute('SELECT * FROM persons WHERE salesrep=%s', 'John Doe')
print( "all persons" )
print( cursor_1.fetchall() ) # 显示出的是cursor_2游标查询出来的结果
print( "John Doe" )
print( cursor_2.fetchall() ) # 不会有任何结果为了避免上述的问题可以使用以下两种方式：
1. 创建多个连接来保证多个查询可以并行执行在不同连接的游标上
2. 使用fetchall方法获取到游标查询结果之后再执行下一个查询，
示例如下：
c1.execute('SELECT ...')
c1_list = c1.fetchall()
c2.execute('SELECT ...')
c2_list = c2.fetchall()
游标返回字典变量
上述例子中游标获取的查询结果的每一行为元组类型，
可以通过在创建游标时指定as_dict参数来使游标返回字典变量，
字典中的键为数据表的列名
import pymssql
# 可以进入 connect 方法里面查看更多参数
conn = pymssql.connect(server, user, password, database) # 可以在连接时指定参数 as_dict=True
cursor = conn.cursor(as_dict=True) # 也可以在创建游标时指定参数 as_dict=True
cursor.execute('SELECT * FROM persons WHERE salesrep=%s', 'John Doe')
for row in cursor:
print("ID=%d, Name=%s" % (row['id'], row['name']))
cursor.close()
conn.close()
使用with语句（上下文管理器）
可以通过使用with语句来省去显示的调用close方法关闭连接和游标
import pymssql
with pymssql.connect(server, user, password, database) as conn:
with conn.cursor(as_dict=True) as cursor:
cursor.execute('SELECT * FROM persons WHERE salesrep=%s', 'John Doe')
for row in cursor:
print("ID=%d, Name=%s" % (row['id'], row['name']))简单使用示例：
import pymssql
conn = pymssql.connect(server, user, password, database)
cur=conn.cursor()
cur.execute('select top 5 * from [dbo].[Dim_Area]')
# 如果 update/delete/insert 记得要 conn.commit() ，否则数据库事务无法提交
# 如果没有指定autocommit属性为True的话就需要调用commit()方法
# conn.commit()
print (cur.fetchall())
cur.close()
cursor = conn.cursor()
cursor.executemany(
"INSERT INTO persons VALUES (%d, %s, %s)",
[(1, 'John Smith', 'John Doe'),
(2, 'Jane Doe', 'Joe Dog'),
(3, 'Mike T.', 'Sarah H.')])
# 如果没有指定autocommit属性为True的话就需要调用commit()方法
conn.commit()
# 查询操作
cursor.execute('SELECT * FROM persons WHERE salesrep=%s', 'John Doe')
row = cursor.fetchone()
while row:
print("ID=%d, Name=%s" % (row[0], row[1]))
row = cursor.fetchone()
# 也可以使用for循环来迭代查询结果
# for row in cursor:
# print("ID=%d, Name=%s" % (row[0], row[1]))
# 关闭连接
conn.close()示例代码：
#!/usr/bin/python3
# -*- coding: utf-8 -*-
import pymssql
db_host = '192.168.0.5'
db_port = '9526'
db_user = 'test'
db_pwd = 'test'
db_name = 'TestDB'
tb_name = 'TestTB'
class SqlServerOperate(object):
def __init__(self, server, port, user, password, db_name):
self.server = server
self.port = port
self.user = user
self.password = password
self.db_name = db_name
pass
def __del__(self):
# 其他地方不关闭连接，可以在这个地方统一关闭
# self.conn.close()
pass
def __get_connect(self):
self.conn = pymssql.connect(
server=self.server,
port=self.port,
user=self.user,
password=self.password,
database=self.db_name,
as_dict=True,
charset="utf8"
)
cur = self.conn.cursor()
if not cur:
raise (NameError, "连接数据库失败")
else:
return cur
def exec_query(self, sql):
cur = self.__get_connect()
cur.execute(sql)
result_list = list(cur.fetchall())
self.conn.close() # 查询完毕后必须关闭连接
# 使用with语句（上下文管理器）来省去显式的调用close方法关闭连接和游标
print('****************使用 with 语句******************')
with self.__get_connect() as cur:
cur.execute(sql)
val = list(cur.fetchall()) # 把游标执行后的结果转换成 list
print(val)
return result_list
def exec_non_query(self, sql):
cur = self.__get_connect()
cur.execute(sql)
self.conn.commit()
self.conn.close() # 完毕后必须关闭连接
def test():
ms = SqlServerOperate(db_host, db_port, db_user, db_pwd, db_name)
sql_string = "select * from SpiderItem where ResourceType = 20"
temp_result_list = ms.exec_query(sql_string)
for i in temp_result_list:
print(i)
pass
if __name__ == "__main__":
test()
pass
改进版：
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author :
# @File : mongoDBoperate.py
# @Software : PyCharm
# @description :
import pymssql
db_host = '192.168.0.5'
db_port = '9526'
db_user = 'test'
db_pwd = 'test'
db_name = 'TestDB'
tb_name = 'TestTB'
class SqlServerOperate(object):
def __init__(self, server, port, user, password, db_name, as_dict=True):
self.server = server
self.port = port
self.user = user
self.password = password
self.db_name = db_name
self.conn = self.get_connect(as_dict=as_dict)
pass
def __del__(self):
self.conn.close()
def get_connect(self, as_dict=True):
conn = pymssql.connect(
server=self.server,
port=self.port,
user=self.user,
password=self.password,
database=self.db_name,
as_dict=as_dict,
charset="utf8"
)
return conn
def exec_query(self, sql):
cur = self.conn.cursor()
cur.execute(sql)
result_list = list(cur.fetchall())
cur.close()
# 使用with语句（上下文管理器）来省去显式的调用close方法关闭连接和游标
# print('****************使用 with 语句******************')
# with self.get_connect() as cur:
# cur.execute(sql)
# result_list = list(cur.fetchall()) # 把游标执行后的结果转换成 list
# # print(result_list)
return result_list
def exec_non_query(self, sql, params=None):
cur = self.conn.cursor()
# cur.execute(sql, params=params)
cur.execute(sql, params=params)
self.conn.commit()
cur.close()
def exec_mutil_sql(self, sql, data_list):
"""
执行一次 sql, 批量插入多条数据
:param sql: 参数用 %s 代替 : insert into table_name(col1, col2, col3) values(%s, %s, %s)
:param data_list: list类型, list中每个元素都是元组
:return:
"""
cur = self.conn.cursor()
cur.executemany(sql, data_list)
self.conn.commit()
cur.close()
def test():
ms = SqlServerOperate(db_host, db_port, db_user, db_pwd, db_name)
sql_string = "select * from SpiderItem where ResourceType = 20"
temp_result_list = ms.exec_query(sql_string)
for i in temp_result_list:
print(i)
pass
if __name__ == "__main__":
test()
pass
相关文章