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']))