简介
pymysql是一个客户端，本文进行一个简单封装，主要是结构化和批量操作
代码
import pymysql class MysqlCli: def __init__(self,host,user,password,database,port=3306,charset='utf8',use_unicode=True): """ 初始化 :param host:主机名 :param user: 用户 :param password: 密码 :param database: 数据库 :param port: 端口，默认3306 :param charset: 字符集，默认utf8 :param use_unicode: 使用unicode，默认True """ try: self.connect=pymysql.connect( host=host, user=user, password=password, database=database, port=port, charset=charset, use_unicode=use_unicode ) self.cursor = self.connect.cursor() print(f"连接数据库{host} {database}成功!!!") except Exception as e: print(f"init error: {e}") def close(self): self.cursor.close() self.connect.close() # 无参 def execute(self, sql): try: self.cursor.execute(sql) self.connect.commit() except Exception as e: self.connect.rollback() raise Exception(f"execute: {e}") # 增删改（结构化） def operate(self,sql,params=None): try: self.cursor.executemany(sql,params) self.connect.commit() except Exception as e: print(f"operate error: {e}") self.connect.rollback() # 查询（结构化） def select(self,sql,params=None): try: tmp_cursor = self.connect.cursor(pymysql.cursors.DictCursor) tmp_cursor.execute(sql,params) res = tmp_cursor.fetchall() tmp_cursor.close() return res except Exception as e: print(f"select error: {e}")
测试
上面代码放到了包sql_helper的helper.py中
from sql_helper.helper import MysqlCli import uuid if __name__ == '__main__': my_cli = MysqlCli(host="10.28.144.100", user="root", password="123456", database="peewee_learn") # 增加 sql= "insert into users(id,name,age,gender,`desc`,time)VALUES(%s,%s,%s,%s,%s,%s)" print(sql) my_cli.operate(sql,[(str(uuid.uuid4()),'lady_killer',22,'1',"something","2023-01-25 19:39:37")]) # 参数化查询 sql = "select * from users where age = %s" res = my_cli.select(sql,18) print(res) my_cli.close()
结果截图：
可以结合Python-logging详解（彩色日志扩展，多进程安全等），添加日志。如果是并发情况下，不能仅使用一个客户端，可以考虑连接池。
