pymysql 数据库操作

from pymysql import *class OCR_DB(object):def __init__(self, user, password, database, port=3306, host='localhost'):self.host = hostself.port = portself.user = userself.password = passwordself.database = databaseself.conn = self.create_conn()  # 光标if not self.conn:exit()self.cursor = self.conn.cursor(cursor=cursors.DictCursor)   # 查询结果以字典方式返回def __del__(self):"""程序退出 释放资源"""try:self.conn.close()self.cursor.close()except:passdef create_conn(self):"""创建与数据库连接"""try:# cursorclass=cursors.DictCursor  字典数据返回conn = connect(host=self.host, port=self.port, user=self.user, password=self.password, database=self.database, charset='utf8', cursorclass=cursors.DictCursor)return connexcept:return Nonedef item_getdata(self, serial_number=None, item_value=None, item=None, db_name=None):""" 通过自定义属性查询值   唯一码  属性值, 自定义查询条件项, 数据表名serial_number: 唯一码,item_value: 查询字段的值,item: 字段,db_name: 数据库名"""try:# 多条数据查询  获取数据集if db_name=="as400_sync" and item=="ticket_bar":getAS400data = []if isinstance(item_value, list):sql = "select * from {} where {}=%s".format(db_name, item)for value in item_value:try:self.cursor.execute(sql, value)as400Data = self.cursor.fetchall()getAS400data.extend(as400Data)except Exception as e:return getAS400data.append("查询异常: %s" % e)return getAS400data# 单条数据获取elif isinstance(item_value, int):try:sql = """select * from {} where {}={}""".format(db_name, item, item_value)self.cursor.execute(sql)as400Data = self.cursor.fetchall()return as400Dataexcept Exception as e:return "数据查询异常: %s" % e# 数据记录表查询else:if item == None:if item_value==None:if serial_number==None:sql = """select * from {}""".format(db_name)   # 没有参数获取数据表全部数据else:sql = ' select * from {} where serial_number={} '.format(db_name, serial_number)   # item_value item 为空时只按唯一码查询else:if db_name==None:if isinstance(item_value, str):   # ocr扫描记录查询  判断输入值为字符串item = 'RMC_bar'            # 默认查询的字段db_name= 'ocr_scan_log'# sql = " select * from {} where serial_number={} and {}='{}' ".format(db_name, serial_number, item, item_value)elif isinstance(item_value, int):   # 高拍仪数据查询  判断职位int整型item = 'ticket_bar'    # 默认查询的字段db_name = 'out_ticket'  # 默认操作表# sql = " select * from {} where serial_number={} and {}='{}' ".format(db_name, serial_number, item, item_value)else:if isinstance(item_value, str):item = 'RMC_bar'            # 默认查询的字段elif isinstance(item_value, int):   # 高拍仪数据查询  判断职位int整型print("部番唯一码查询数据信息")item = 'ticket_bar'    # 默认查询的字段sql = " select * from {} where serial_number={} and {}='{}' ".format(db_name, serial_number, item, item_value)else:if serial_number==None:item = "ticket_bar"db_name = "out_ticket"sql = " select * from {} where {}='{}' ".format(db_name, item, item_value)else:sql = " select * from {} where serial_number={} and {}='{}' ".format(db_name, serial_number, item, item_value)# print(sql)self.cursor.execute(sql)outTicket_getdata = self.cursor.fetchall()return outTicket_getdataexcept Exception as e:print("查询异常: ", e)def insert_one_scanLog(self, RMC_bar, BOX, serial_number, db_name='ocr_scan_log'):""" 单条数据插入入 OCR扫描数据 """sql = " insert into {} values %s ".format(db_name)try:self.cursor.execute(sql, [(0, RMC_bar, BOX, serial_number, '')])self.conn.commit()except Exception as e:print("输入插入有异常", e)self.conn.rollback()   # 异常回转, 所有数据插入无效def insert_Group_List(self, insert_dataList, db_name='ocr_scan_log'):""" 插入多组数据 列表方式插入insert_dataList: 插入的数据 列表套元组, 多条为多个元组db_name: 数据表名"""val = '%s, ' * (len(insert_dataList[0]) - 1) + '%s'sql = f'insert into {db_name} values ({val})'try:# self.cursor.execute(sql, insert_dataList)# 多条数据插入# 插入数据格式  [(0, '99yyy', 779, 987656787890, ''), (0, 'QWE8890', 990, 987654456789, '')]self.cursor.executemany(sql, insert_dataList)self.conn.commit()except Exception as e:print("输入插入有异常", e)self.conn.rollback()  # 异常回转, 所有数据插入无效def insert_Group_dict(self, insert_dataDict=None, serial_number=None, db_name='ocr_scan_log'):""" 字典方式插入数据 多组数据插入  防SQL注入插入数据insert_dataDict: 插入的数据, 单条数据为字典, 多条插入为列表套字典,serial_number: 唯一码,db_name: 数据表名"""# 检查数据是否有重复的try:de_weight = self.item_getdata(db_name=db_name, serial_number=serial_number)except:de_weight = ()if not de_weight:   # 如有重复数据 停止插入# 这里db_name和insert_dataDict，其中insert_dataDict是一个字典，写插入数据库的代码# 获取插入数据的头部数据 KEYS  列名# 传进来参数为字典, 当insert_dataDict为字典时 为单个数据插入if isinstance(insert_dataDict, dict):cols = ", ".join('`{}`'.format(k) for k in insert_dataDict.keys())# 获取值val_cols = ', '.join('%({})s'.format(k) for k in insert_dataDict.keys())a = [insert_dataDict]  # 拼出要插入的内容# 数据为列表嵌套字典时为多字典/多条数据插入elif isinstance(insert_dataDict, list):cols = ", ".join('`{}`'.format(k) for k in insert_dataDict[0].keys())# 获取值val_cols = ', '.join('%({})s'.format(k) for k in insert_dataDict[0].keys())a = insert_dataDict# 获取插入SQL 命令 insert_dataDictsql = "insert ignore into {}(%s) values(%s)".format(db_name)res_sql = sql % (cols, val_cols)try:self.cursor.executemany(res_sql, a)  # 将字典列表a传入self.cursor.connection.commit()# self.conn.commit()except Exception as e:print("字典数据插入异常: ", e)self.conn.rollback()   # 回滚 插入数据无效else:print("数据重复插入")def update_db_data(self, serial_number=None, con_item='RMC_bar', con_itemValue=None, item="error", item_value=None, result=0, db_name=None):"""数据更新   部番号, 唯一码, 修改项默认为error, 修改值, 数据表名con_item: 判断条码字段,内容  默认RMC_barcon_itemValue: 字段条件serial_number: 唯一码,item: 修改的字段,item_value: 修改的值,db_name: 数据表名  没有指定操作数据表对象, 默认对 ocr_scan_log, out_ticket两个表进行操作"""# 不指定操作表  默认两个表全部进行操作if db_name==None:db_name = ['ocr_scan_log', 'out_ticket']try:for db_obj in db_name:# 列表方式 多个参数同时更新if isinstance(item_value, list):if db_obj=='out_ticket':sql = """ update {} set {}=%s, result={} where serial_number=%s and {}=%s """.format(db_obj, item, result, con_item)elif db_obj=='ocr_scan_log':sql = """ update {} set {}=(%s) where serial_number=(%s) and {}=(%s) """.format(db_obj, item, con_item)self.cursor.executemany(sql, item_value)else:if db_obj=='out_ticket':sql = """ update {} set {}='{}',result={} where serial_number={} and {}='{}' """.format(db_obj, item, item_value, result, serial_number, con_item, con_itemValue)elif db_obj=='ocr_scan_log':sql = """ update {} set {}='{}' where serial_number={} and {}='{}' """.format(db_obj, item, item_value, serial_number, con_item, str(con_itemValue))self.cursor.execute(sql)self.cursor.connection.commit()except Exception as e:print(e)self.conn.rollback()# 指定表操作对象else:if item=="error" and db_name=="out_ticket":sql = """ update {} set {}='{}',result={} where {}={} and {}='{}' """.format(db_name, item, item_value, result, 'serial_number', serial_number, con_item, str(con_itemValue))else:sql = """ update {} set {}='{}' where {}={} and {}='{}' """.format(db_name, item, item_value, 'serial_number', serial_number, con_item, str(con_itemValue))try:self.cursor.execute(sql)self.conn.commit()except:self.conn.rollback()   # 异常触发回滚 操作无效def create_view(self, table_name, new_viewName="view", field_name=["*"]):""" table_name: 目标表名,new_viewName: 视图名,field_name: 字段名 [列表传参数, 默认*]"""cols = ", ".join('`{}`'.format(k) for k in field_name)  # 获取字段信息sql = """ create view {}  as  select {} from {} """.format(new_viewName, cols, table_name)self.cursor.execute(sql)self.cursor.connection.commit()def if_empty(self, con_item_dict=None,  db_name="ocr_scan_log"):"""db_name: 数据表名  无参数为查询表是否有数据"""if con_item_dict:if isinstance(con_item_dict, dict):sql = """select * from %s where {}""" % db_namecols_str = " and ".join('{}="{}"'.format(k, values) for k, values in con_item_dict.items())sql = sql.format(cols_str)else:sql = """select * from %s""" % db_name# print("执行语句>> ", sql)self.cursor.execute(sql)check_data = self.cursor.fetchall()if check_data:return Trueelse:return Falseif __name__=='__main__':ocr_db_obj = OCR_DB(user='root', password='root', database='ocr')ocr_db_obj.if_empty({"RMC_bar": "D2421231-+-P"}, db_name="as400_sync")# ocr_db_obj.insert_Group_dict({'RMC_bar': "tyuio456789", 'BOX': 12, 'serial_number': 34567890, 'error': ''}, 'ocr_scan_log')# ocr_db_obj.insert_one_scanLog("ghjsf56789", 13, 4567890,  'ocr_scan_log')# ocr_db_obj.insert_Group_dict( serial_number=20210419172268, insert_dataDict=[{'RMC_bar': 'D2421231-+-P', 'BOX': 1, 'serial_number': 20210419172268, 'error': ''}, {'RMC_bar': 'M0AJ5891A-+-P', 'BOX': 1, 'serial_number': 20210419172268, 'error': ''}, {'RMC_bar': 'M0AJ5892A-+-P', 'BOX': 1, 'serial_number': 20210419172268, 'error': ''}, {'RMC_bar': 'M0AJ1235D-+-P', 'BOX': 1, 'serial_number': 20210419172268, 'error': ''}, {'RMC_bar': 'D2417943-+-P', 'BOX': 1, 'serial_number': 20210419172268, 'error': ''}, {'RMC_bar': 'D2417944A-+-P', 'BOX': 1, 'serial_number': 20210419172268, 'error': ''}, {'RMC_bar': 'D2417984C-+-P', 'BOX': 1, 'serial_number': 20210419172268, 'error': ''}, {'RMC_bar': 'D2415841C-+-P', 'BOX': 3, 'serial_number': 20210419172268, 'error': ''}, {'RMC_bar': 'D1497982C-+-P', 'BOX': 1, 'serial_number': 20210419172268, 'error': ''}])  # D2415841C-+-P# ocr_db_obj.create_view(table_name='out_ticket', new_viewName="ocr")# ocr_db_obj.if_empty()# a = ocr_db_obj.item_getdata(item_value="M0AJ5891A-+-P", serial_number=20210420164086)# item_value = [['3242136 工厂编号异常', 20210421152693, 'M0AJ1235D-+-P'],#               ['3242136 工厂编号异常', 20210421152693, 'M0AJ5892A-+-P'],#               ['3242136 工厂编号异常', 20210421152693, 'M0AJ5891A-+-P'],#               ['3242136 工厂编号异常', 20210421152693, 'D2421231-+-P']]# ocr_db_obj.update_db_data(item_value=item_value, result=0)