0
点赞
收藏
分享

微信扫一扫

Python通用的创建数据库连接方法


json文件参数

'''
def __init__(self, host=None, user=None, password="",
database=None, port=0, unix_socket=None,
charset='', sql_mode=None,
read_default_file=None, conv=None, use_unicode=None,
client_flag=0, cursorclass=Cursor, init_command=None,
connect_timeout=10, ssl=None, read_default_group=None,
compress=None, named_pipe=None,
autocommit=False, db=None, passwd=None, local_infile=False,
max_allowed_packet=16*1024*1024, defer_connect=False,
auth_plugin_map=None, read_timeout=None, write_timeout=None,
bind_address=None, binary_prefix=False, program_name=None,
server_public_key=None):
'''
#例:
{"host":"127.0.0.1",
"user":"root",
"password":"******",
"database":"newtest",
"port":3306,
"charset":"utf8"
}

import pymysql
import json
class BaseDao():
'''
通用的创建数据库连接方法
'''
def __init__(self,configfile='mysql.json'):
self.config=json.load(open(configfile))
self.connection=None
self.cursor=None
pass
def getConnection(self):
if self.connection!=None:
return self.connection
try:
self.connection=pymysql.connect(** self.config)
self.connection.autocommit(0)
except Exception as e:
print("数据库连接失败,请检查配置参数")
pass
print("数据库连接成功")
return self.connection
pass
def execute(self,sql,params=None):
'''
封装执行语句的方法
:param sql: 语句
:param params: 参数列表
:return:
'''
result=0
try:
self.cursor=self.getConnection().cursor()
if params: #if None!=None if None:False
result=self.cursor.execute(sql,params)
else:
result=self.cursor.execute(sql)
except Exception as e:
print(e)
return result
def fetch(self):
if self.cursor:
return self.cursor.fetchall()
pass
# 处理事务管理
def commit(self):
if self.connection:
self.connection.commit()
pass
# 回滚
def rollback(self):
if self.connection:
self.connection.rollback()
pass
# 关闭连接
def close(self):
if self.connection:
self.connection.close()
pass


举报

相关推荐

0 条评论