python操作mysql基礎增刪改查的簡單封裝
阿新 • • 發佈:2021-01-21
傳入物件實現增刪改查,無需自己手動拼接
例
增加:
# 傳入的obj
insert_obj = { # 增加 ,select和where可為空 'id': 111, 'order': 111, 'order_id': 111, 'rate_star': 111, }
刪除:
# 傳入的obj
delete_obj = { # 刪除 select 可省去 # 刪除條件欄位 'order_id': 111, 'rate_star': 111,'where_list': ['order_id', 'rate_star'] }
更改:
# 傳入的obj
update_obj = { # 更新 select可省去, # 待更新欄位 'order_id': 60263167670871, ... 'id': 60263167670871, # 條件欄位 'rate_star': 1111, 'order_id': 60263167670871, 'where_list': ['order_id', 'rate_star'] }
查詢:
# 傳入的obj
select_obj = { # 查詢 # 條件欄位 'order_id': 111, 'rate_star': 111, 'where_list': ['order_id', 'rate_star'], # 待查詢欄位 'select_list': ['id', 'star'] }
程式碼:
# *coding:utf-8 * import json import pymysqlclass db_settings: ''' 例: db_obj = { 'field1': 'field1', 'field2': 'field2', 'where_field1':'where_field1', 'where_field2':'where_field1', 'select_list': ['select_field1', 'select_field2'], # select查詢欄位 'where_list': ['where_field1', 'where_field2'] # where條件欄位 } ''' def __init__(self, db=None, user=None, password=None, host='localhost', port=3306, charset='utf8'): self.conn_mysql = pymysql.Connect( database=db, user=user, password=password, host=host, port=port, charset=charset, ) def __del__(self): self.conn_mysql.close() def db_selete(self, *args, **kwargs): # 獲取資料欄位 # 整理出sql # 呼叫db table = args[0] where_fields = '' data = kwargs.get('data') where_list = data.get('where_list') select_list = data.get('select_list') if where_list !=None: del data['where_list'] if select_list != None: del data['select_list'] for k, v in data.items(): if k in where_list: if where_fields == '': where_fields += f"{k}='{v}'" else: where_fields += f"and {k}='{v}'" fields = ','.join(select_list) cursor = self.conn_mysql.cursor() sql = f"""select {fields} from {table} where {where_fields}""" cursor.execute(sql) result = cursor.fetchall() return result def db_update(self, *args, **kwargs): table = args[0] fields = '' where_fields = '' data = kwargs.get('data') where_list = data.get('where_list') select_list = data.get('select_list') if where_list != None: del data['where_list'] if select_list != None: del data['select_list'] for k, v in data.items(): if k in where_list: if where_fields == '': where_fields += f"{k}='{v}'" else: where_fields += f"and {k}='{v}'" else: if fields == '': fields += f"{k}='{v}'" else: fields += f", {k}='{v}'" # 呼叫sql cursor = self.conn_mysql.cursor() sql = f"""update {table} set {fields} where {where_fields}""" try: cursor.execute(sql) self.conn_mysql.commit() except Exception as e: print(e) self.conn_mysql.rollback() def db_insert(self, *args, **kwargs): table = args[0] fields = '' where_fields = '' data = kwargs.get('data') where_list = data.get('where_list') select_list = data.get('select_list') if where_list != None: del data['where_list'] if select_list != None: del data['select_list'] num = 0 for k, v in data.items(): if num == 0: where_fields += f"{k}" fields += f"'{v}'" else: where_fields += f", {k}" fields += f", '{v}'" num += 1 cursor = self.conn_mysql.cursor() sql = f"""insert into {table} ({where_fields}) values({fields})""" try: cursor.execute(sql) self.conn_mysql.commit() except Exception as e: print(e) self.conn_mysql.rollback() def db_delete(self, *args, **kwargs): table = args[0] fields = '' where_fields = '' data = kwargs.get('data') where_list = data.get('where_list') select_list = data.get('select_list') if where_list != None: del data['where_list'] if select_list != None: del data['select_list'] for k, v in data.items(): if fields == '': fields += f"{k}='{v}'" else: fields += f", {k}='{v}'" if k in where_list: if where_fields == '': where_fields += f"{k}='{v}'" else: where_fields += f"and {k}='{v}'" cursor = self.conn_mysql.cursor() sql = f"""delete from {table} where {where_fields}""" try: cursor.execute(sql) self.conn_mysql.commit() except Exception as e: print(e) self.conn_mysql.rollback() if __name__ == '__main__': # 操作 local_class = db_settings(db='db', user='root', password='123', host='host') selete_obj = { 'order_id': 60263167670871, 'rate_star': json.dumps({"111":111}), 'select_list': ['order', 'test'], 'where_list': ['order_id', 'rate_star'] } local_class.db_selete('shopee_customer', data=selete_obj)