当前位置 博文首页 > 小白的成长记录:Python 使用with管理数据库连接 MySQL/MSSQL/SQ
MySQL和SQLServer、SQLite都大致相同
import pymssql
MSSQL_AUTH = {'host': '127.0.0.1', 'user': 'sa', 'password': 'sa.123', 'database': 'DevDatabase'}
MYSQL_AUTH = {'host': '127.0.0.1', 'user': 'root', 'password': 'Root.123', 'db': 'DevDatabase'}
class MSSQL:
def __init__(self, auth: dict = None, charset="utf8", as_dict=True):
auth = auth if auth else MSSQL_AUTH
self.conn = pymssql.connect(**auth, charset=charset, as_dict=as_dict)
self.cur = self.conn.cursor() # 创建游标
""" MySQL替换两行代码
self.conn = pymysql.connect(**auth, charset=charset)
self.cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
"""
def exec_insert(self, table, data: dict):
"""
:param table: TabelName
:param data: {'FieldName':'Value',}
:return: lastrowid
"""
sql = "INSERT INTO {} ({}) VALUES ({})"
key_list, val_list = [], []
for k, v in data.items():
key_list.append(k)
val_list.append(v)
# SQLite数据库这里的's%'要换成'?'
sql = sql.format(table, ','.join(key_list), ','.join(['%s'] * len(val_list)))
self.cur.execute(sql, tuple(val_list))
self.conn.commit() # 提交
return self.cur.lastrowid
def execute(self, sql, commit=False):
"""
:param sql: 执行查询或更新语句
:param commit: 是否提交,执行更新语句时=True
:return: 查询结果集/None
"""
self.cur.execute(sql)
if commit:
self.conn.commit()
else:
return self.cur.fetchall()
def get_value(self, table, eq=0, **kwargs):
"""
:param table: 表名
:param eq: 数据下标索引,负数不使用索引
:param kwargs: 依据,条件
:return: 结果
"""
condition = 'and'.join([f"{k}='{v}'" for k, v in kwargs.items()])
self.cur.execute(f"SELECT * FROM {table} WHERE {condition}")
if eq >= 0:
return self.cur.fetchall()[eq] # 需要的话,可以添加安全判断,防止下标越界
return self.cur.fetchall()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.cur.close() # 关闭游标
self.conn.close() # 关闭链接
if __name__ == '__main__':
with MSSQL() as db:
res = db.execute("SELECT TOP(100) * FROM TelRecord")
print(res)
cs