当前位置：首页 > news >正文

python连接sqlserver，封装操作

news 2025/8/4 11:37:33

1封装

# 导入Flask类
import pymssql
import tracebackclass Mssql(object):# 连接库def base(database):connect = pymssql.connect(user='sa',password='123456',database=f'{database}',charset='utf8',as_dict=True)if connect:print("数据库连接成功！")else:print("链接失败！")return connect# 查询数据def query(connect,sql):try:cursor = connect.cursor()cursor.execute(sql)req = cursor.fetchall()except Exception:#打印错误信息traceback.print_exc()return "出错了"# print(res)return req# 新增,table表，data数据def inserts(connect, table, data):# 组装数据name = ''val = ''for v1 in data:if name == "":name = f"{v1}"val = f"'{data[v1]}'"else:name = f"{name},{v1}"val = f"{val},'{data[v1]}'"insert = f"INSERT INTO {table}({name}) VALUES({val})"print(insert)try:cursor = connect.cursor()cursor.execute(insert)connect.commit()except Exception:#打印错误信息traceback.print_exc()print(str(Exception))return "出错了"return 1# 修改,table表，data数据，where条件def updates(connect, table, data, where):# 组装数据sql = ''for v1 in data:if sql == "":sql = f"{v1}='{data[v1]}'"else:sql = f"{sql},{v1}='{data[v1]}'"update = f"UPDATE {table} set {sql} where {where}"print(update)try:cursor = connect.cursor()cursor.execute(update)connect.commit()except Exception:#打印错误信息traceback.print_exc()return "出错了"return 1#删除,table表，where条件def deletes(connect, table, where):delete = f"delete from {table} where {where}"print(delete)try:cursor = connect.cursor()cursor.execute(delete)connect.commit()except Exception:#打印错误信息traceback.print_exc()return "出错了"return 1# 关闭连接def close(conn):conn.close()

2使用

from configs.Mssql import Mssql
database = "SD65512_Sample"def query():conn = Mssql.base(database)req = Mssql.query(conn, "SELECT billid FROM s_salechild")print(req)return 1def add():conn = Mssql.base(database)data = {'id': 1}req = Mssql.inserts(conn, "dade", data)print(req)return 1def update():conn = Mssql.base(database)data = {'dade': 888}Mssql.updates(conn, "dade", data, "id=1")return 1def delete():conn = Mssql.base(database)Mssql.deletes(conn, "dade", "id=1")return 1# 关闭连接
def close():conn = Mssql.base(database)conn.close()# 测试
# query()
# add()
# update()
delete()