?

Python連接SQL Server數據庫 - pymssql使用基礎：https://www.cnblogs.com/baiyangcao/p/pymssql_basic.html

廖雪峰官網 之 Python 訪問數據庫（SQLLite / MySQL / SQLAlchemy）

?

pymssql?examples ：http://pymssql.org/en/stable/pymssql_examples.html

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=Truecursor.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 pymssqlwith 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 pymssqlconn = 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 pymssqldb_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 = serverself.port = portself.user = userself.password = passwordself.db_name = db_namepassdef __del__(self):# 其他地方不關閉連接，可以在這個地方統一關閉# self.conn.close()passdef __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 curdef 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()) # 把游標執行后的結果轉換成 listprint(val)return result_listdef 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)passif __name__ == "__main__":test()pass

?

改進版：

#!/usr/bin/python3 # -*- coding: utf-8 -*- # @Author : # @File : mongodboperate.py # @Software : PyCharm # @description :import pymssqldb_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 = serverself.port = portself.user = userself.password = passwordself.db_name = db_nameself.conn = self.get_connect(as_dict=as_dict)passdef __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 conndef 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_listdef 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)passif __name__ == "__main__":test()pass

?

?

?

?

?

?

總結

以上是生活随笔為你收集整理的Python操作MSSQL的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。