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