安装:pip install DBUtils==1.3
代码如下:
"""连接池操作pymysql""" import pymysql from DBUtils.PooledDB import PooledDB class MysqlPool(object): def __init__(self): mysql_deploy = { "maxusage": None, "blocking": True, "mincached": 2, "maxcached": 5, "maxshared": 3, "maxconnections": 10, # 连接池允许的最大连接数 "creator": pymysql, "host": "xxx", "port": 3306, "user": "xxx", "password": "xxx", "database": "xxxl" } self.POOL = PooledDB(**mysql_deploy) def connect(self): """ 创建连接 :return: """ conn = self.POOL.connection() cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) return conn, cursor def connect_close(self, conn, cursor): """ 关闭连接 :param conn: :param cursor: :return: """ cursor.close() conn.close() def fetch_one(self, sql_): """ 查单条数据 :param sql_: :return: """ conn, cursor = self.connect() cursor.execute(sql_) result = cursor.fetchone() self.connect_close(conn, cursor) return result def fetch_all(self, sql_): """ 批量数据 :param sql_: :return: """ conn, cursor = self.connect() cursor.execute(sql_) db_list = cursor.fetchall() self.connect_close(conn, cursor) return db_list def delect(self, sql_): conn, cursor = self.connect() row = cursor.execute(sql_) conn.commit() self.connect_close(conn, cursor) return row if __name__ == '__main__': mysql = MysqlPool() sql = "DELETE FROM community.user_relation WHERE user_id=3725 OR to_user_id=3725;" data = mysql.delect(sql) print(data)
基本翻译
adj. 三十的
num. 罗马数字30