python 對MySQL操作進行封裝
后面直接拿來用就ok
import pymysql
class MysqlManager(object):
'''mysql管理器'''
def __init__(self, db, user, passwd, host='localhost', port=3306, charset='utf8'):
'''初始化數據庫'''
self.__db = db
self.__user = user
self.__passwd = passwd
self.__host = host
self.__port = port
self.__charset = charset
self.__connect = None
self.__cursor = None
def _connect_db(self):
"""
dbManager._connect_db()
連接數據庫
"""
params = {
"db": self.__db,
"user": self.__user,
"passwd": self.__passwd,
"host": self.__host,
"port": self.__port,
"charset": self.__charset
}
self.__connect = MySQLdb.connect(**params)
self.__cursor = self.__connect.cursor()
def _close_db(self):
'''
dbManager._close_db()
關閉數據庫
'''
self.__cursor.close()
self.__connect.close()
def insert(self, table, insert_data):
'''
dbManager.insert(table, insert_data)
添加數據到數據庫
str -> table 為字符串
[{},{}] -> 為列表中嵌套字典類型
'''
# 用戶傳入數據字典列表數據，根據key, value添加進數據庫
# 連接數據庫
self._connect_db()
try:
for data in insert_data:
# 提取插入的字段
key = ','.join(data.keys())
# 提取插入的值
values = map(self._deal_values, data.values())
insert_data = ', '.join(values)
# 構建sql語句
sql = "insert into {table}({key}) values ({val})".format(table=table, key=key, val=insert_data)
self.__cursor.execute(sql)
self.__connect.commit()
except Exception as error:
print error
finally:
self._close_db()
def delete(self, table, condition):
'''
dbManager.delete(table, condition)
刪除數據庫中的數據
str -> table 字符串類型
dict -> condition 字典類型
'''
self._connect_db()
# 處理刪除的條件
condition_list = self._deal_values(condition)
condition_data = ' and '.join(condition_list)
# 構建sql語句
sql = "delete from {table} where {condition}".format(table=table, condition=condition_data)
self.__cursor.execute(sql)
self.__connect.commit()
self._close_db()
def update(self, table, data, condition=None):
"""
dbManager.update(table, data, [condition])
更新數據
str -> table 字符串類型
dict -> data 字典類型
dict -> condition 字典類型
"""
self._connect_db()
# 處理傳入的數據
update_list = self._deal_values(data)
update_data = ",".join(update_list)
# 判斷是否有條件
if condition is not None:
# 處理傳入的條件
condition_list = self._deal_values(condition)
condition_data = ' and '.join(condition_list)
sql = "update {table} set {values} where {condition}".format(table=table, values=update_data, condition=condition_data)
else:
sql = "update {table} set {values}".format(table=table, values=update_data)
print sql
self.__cursor.execute(sql)
self.__connect.commit()
self._close_db()
def get(self, table, show_list, condition=None, get_one=False):
"""
dbManager.get(table, show_list, [condition, get_one]) -> tupe
獲取數據 返回一個元祖
str -> table 字符串類型
list -> show_list 列表類型
dict -> condition 字典類型
boolean -> get_one 布爾類型
"""
self._connect_db()
# 處理顯示的數據
show_list = ",".join(show_list)
sql = "select {key} from {table}".format(key=show_list, table=table)
# 處理傳入的條件
if condition:
condition_list = self._deal_values(condition)
condition_data = 'and'.join(condition_list)
sql = "select {key} from {table} where{condition}".format(key=show_list, table=table, condition=condition_data)
self.__cursor.execute(sql)
# 返回一條數據還是所有數據
if get_one:
result = self.__cursor.fetchone()
else:
result = self.__cursor.fetchall()
self._close_db()
return result
def _deal_values(self, value):
"""
self._deal_values(value) -> str or list
處理傳進來的參數
"""
# 如果是字符串則加上''
if isinstance(value, str):
value = ("'{value}'".format(value=value))
# 如果是字典則變成key=value形式
elif isinstance(value, dict):
result = []
for key, value in value.items():
value = self._deal_values(value)
res = "{key}={value}".format(key=key, value=value)
result.append(res)
return result
else:
value = (str(value))
return value
方式二：去除mysql語句的操作
import pymysql
class MysqlHelper():
def __init__(self,host,port,db,user,passwd,charset='utf8'):
self.host = host
self.port = port
self.db = db
self.user = user
self.password = passwd
self.charset = charset
def open(self):
# 連接數據庫
self.conn = pymysql.connect(host=self.host,port=self.port,db=self.db,
user=self.user,passwd=self.password,charset=self.charset)
# 創建游標對象
self.cursor = self.conn.cursor()
# 關閉
def close(self):
self.cursor.close()
self.conn.close()
# 增加、修改、刪除命令語句
def cud(self,sql,params,msg="操作成功"):
try:
self.open()
# 處理邏輯數據，傳入sql語句以及參數化
self.cursor.execute(sql,params)
# 執行事務
self.conn.commit()
self.close()
# 這樣可以修改輸出的操作成功信息提示
print(msg)
except Exception as e:
self.conn.rollback()
print("錯誤",e)
# 查詢所有數據,多個值
def all(self,sql,params=()):
try:
self.open()
self.cursor.execute(sql,params)
data = self.cursor.fetchall()
self.close()
return data
except Exception as e:
print("錯誤", e)