Python使用sqlserver數據庫

sqlserver數據庫安裝

1.自行安裝

python使用

1.下載模塊

pip install pymssql #2.2.1版本

2.使用流程

參考鏈接：https://blog.csdn.net/lin_strong/article/details/82880806

參數名(類型)	說明
server (str)	數據庫主機
user (str)	用於連接的數據庫用戶
password (str)	用戶的密碼
database (str)	鏈接初始化的數據庫。默認情況下，SQL服務器會選擇設置中特定用戶所對應的默認數據庫。
timeout (int)	用秒表示的查詢超時時間，默認為0(無超時)
login_timeout (int)	用秒表示的連接與登陸超時時間，默認為60
charset (str)	連接到數據庫所使用的字符集
as_dict (bool)	是否每一行作為字典而不是元組返回。你可以使用基於0的索引或者使用名字來訪問列。
host (str)	你想要連接的數據庫主機或實體。如： r’.\SQLEXPRESS’ –本地機器上的SQLEXPRESS實體(僅Windows) r’(local)\SQLEXPRESS’ – 同上(僅Windows) ‘SQLHOST’ – 默認端口上的默認實體(僅Windows) ‘SQLHOST’ – 在freetds.conf中設置的指定端口上的指定實體 (僅Linux/*nix) ‘SQLHOST,1433’ – 指定主機上的指定TCP端口 ’SQLHOST:1433’ – 同上 ’SQLHOST,5000’ – 如果你已經設置了一個實體在端口5000進行監聽 ’SQLHOST:5000’ – 同上 ’.’ (本地主機)默認設置，如果沒有指定host。
appname (str)	設置鏈接使用的應用名
port (str)	連接到服務器所使用的TCP端口號
conn_properties	當鏈接建立時發送給服務器的SQLqueries。可以是一個字符串或者另一類可迭代的字符串組。默認值：見_mssql.connect()
autocommit (bool)	是否使用默認自動提交模式
tds_version (str)	使用的TDS協議版本

警告：

目前，設置timeout或login_timeout會有一個過程級的影響，因為用於實現超時的FreeTDS db-lib API函數是全局效果的。

 

https://blog.csdn.net/lin_strong/article/details/82868160?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522161967253216780264042290%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=161967253216780264042290&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~top_positive~default-1-82868160.first_rank_v2_pc_rank_v29&utm_term=pymssql

#1.導入模塊
import pymssql

#2.建立鏈接：server(ip)，user(用戶名)，password(密碼)，database(數據庫)，autocommit(是否自動提交)，charset(編碼方式:cp36)
conn = pymssql.connect(server="127.0.0.1",user="sa",password="123456",database="sqldbtest",autocommit=True, port='1433',charset='cp936')

#3.獲取游標
# cur = conn.cursor() #結果形式：[(),(),..]
cur = conn.cursor(as_dict=True) #[{},{},..]
if not cur:
    raise (NameError, "連接數據庫失敗")  # 將DBC信息賦值給cur

#4.執行sql語句
#查詢語句執行：此方法可解決中文編碼和sql注入
name = "張三".encode("cp936")#中文字符編碼
s1 = """
select * from test_name where name=%s order by age
"""
cur.execute(s1,(name,))
#獲取結果
#獲取一個
# res = cur.fetchone() #(1, '張三', '男', '18852671307', 18) {'id': 1, 'name': '張三', 'sex': '男', 'phone': '18852671307', 'age': 18}
#獲取所有
# res = cur.fetchall() #[(1, '張三', '男', '18852671307', 18), (2, '李四', '男', '18852671311', 19)] [{'id': 1, 'name': '張三', 'sex': '男', 'phone': '18852671307', 'age': 18}, {'id': 2, 'name': '李四', 'sex': '男', 'phone': '18852671311', 'age': 19}]
#自定義獲取
res = cur.fetchmany(1)#獲取一條數據

#非查詢語句執行
# try:
#     cur.execute(s4)
      # 一次插入多行記錄
      # sql = "INSERT INTO userinfo(username,passwd) VALUES(%s,%s)"
#     cur.executemany(sql,[("tom","123"),("alex",'321')])
#     #如果對數據進行了修改，且在連接時沒有把autocommit設置為True，則需要手動調用commit進行提交修改。
#     # conn.commit()
# except Exception as ex:
#   conn.rollback()
#   raise ex

#5.關閉鏈接
cur.close()
conn.close()

類封裝方法使用

import pymssql

class MSSQL:
    #1.初始化
    def __init__(self,host,user,pwd,db,charset,as_dict=True):
        self.host = host
        self.user = user
        self.pwd = pwd
        self.db = db
        self.charset = charset
        self.as_dict = as_dict
    #2.建立鏈接返回游標
    def __GetConnect(self):
        """
        得到連接信息
        返回: conn.cursor()
        """
        if not self.db:
            raise(NameError,"沒有設置數據庫信息")
        self.conn = pymssql.connect(host=self.host,user=self.user,password=self.pwd,database=self.db,charset=self.charset)
        self.cur = self.conn.cursor(as_dict=self.as_dict)
        if not self.cur:
            raise(NameError,"連接數據庫失敗")
        else:
            return self.cur
    #3.執行查詢語句
    def ExecQuery(self,sql,args=(),restype="all"):
        """
        執行查詢語句
        返回的是一個包含tuple的list，list的元素是記錄行，tuple的元素是每行記錄的字段

        調用示例：
                ms = MSSQL(host="localhost",user="sa",pwd="123456",db="PythonWeiboStatistics")
                resList = ms.ExecQuery("SELECT id,NickName FROM WeiBoUser")
                for (id,NickName) in resList:
                    print str(id),NickName
        """
        cur = self.__GetConnect()
        #執行sql語句
        cur.execute(sql,args)
        #獲取數據
        if restype == "one":
            resList = cur.fetchone() #獲取一條數據
        elif isinstance(restype,int):
            resList = cur.fetchmany(int(restype))#自定義獲取數據
        else:
            resList = cur.fetchall()
        return resList
    #4.執行非查詢語句
    def ExecNonQuery(self,sql,args=()):
        """
        執行非查詢語句

        調用示例：
            cur = self.__GetConnect()
            cur.execute(sql)
            self.conn.commit()
            self.conn.close()
        """
        cur = self.__GetConnect()
        try:
            cur.execute(sql,args)
            self.conn.commit()
        except Exception as e:
            self.conn.rollback()
            print(str(e))

    #5.關閉鏈接
    def close(self):
        # 如果數據打開，則關閉；否則沒有操作
        if self.conn and self.cur:
            self.cur.close()
            self.conn.close()
            return True
        raise Exception("not conn and cur")

#1.建立鏈接
ms = MSSQL(host="localhost",user="sa",pwd="123456",db="sqldbtest",charset="cp936")
#1.查詢
name = "張三".encode("cp936")#中文字符編碼
s1 = """
select * from test_name where id=%s order by age
"""

s2 = """
select * from test_name 
"""
# res = ms.ExecQuery(s2,())
# print(res)

#2.增
# str1 = "json".encode("cp936")
# str2 = "男".encode("cp936")
# phone = "17752671303".encode("cp936")
# age = 18
# a1 = """
# insert into test_name values(%s,%s,%s,%s)
# """
# res = ms.ExecNonQuery(a1,(str1,str2,phone,age))

#3.刪
d1 = """
delete from test_name where name=%s
"""
name = "json".encode("cp936")
res = ms.ExecNonQuery(d1,(name,))
#4.改
# u1 = """
# update test_name  set name='木星' where id=1
# """
# res = ms.ExecNonQuery(u1)
# u2 = """
# update test_name  set name=%s,age = 23 where id=1
# """
# name = "張三".encode("cp936")
# res = ms.ExecNonQuery(u2,(name,))
#關閉
ms.close()

 

上下文使用方法

with pymssql.connect(server, user, password, "tempdb") 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']))