Python操作MSSQL

py3study

发布于 2020-01-09 11:12:15

2.8K0

发布于 2020-01-09 11:12:15

文章被收录于专栏：python3python3

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=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

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2019-08-30 ，如有侵权请联系 cloudcommunity@tencent.com 删除

python