这里只实现了部分功能，可以基于此工具类，加入更多功能，如事物、锁等。

  1. # db.py
  2. import pymysql
  3. from dbutils.pooled_db import PooledDB
  6. class DBHelper(object):
  8.     def __init__(self):
  9.         # TODO 此处配置，可以去配置文件中读取。
  10.         self.pool = PooledDB(
  11.             creator=pymysql,  # 使用链接数据库的模块
  12.             maxconnections=5,  # 连接池允许的最大连接数，0和None表示不限制连接数
  13.             mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
  14.             maxcached=3,  # 链接池中最多闲置的链接，0和None不限制
  15.             blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
  16.             setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
  17.             ping=0,
  18.             # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
  19.             host='127.0.0.1',
  20.             port=3306,
  21.             user='root',
  22.             password='root',
  23.             database='db1',
  24.             charset='utf8'
  25.         )
  27.     def get_conn_cursor(self):
  28.         conn = self.pool.connection()
  29.         cursor = conn.cursor(pymysql.cursors.DictCursor)
  30.         return conn, cursor
  32.     def close_conn_cursor(self, *args):
  33.         for item in args:
  34.             item.close()
  36.     def exec(self, sql, **kwargs):
  37.         conn, cursor = self.get_conn_cursor()
  39.         cursor.execute(sql, kwargs)
  40.         conn.commit()
  42.         self.close_conn_cursor(conn, cursor)
  44.     def fetch_one(self, sql, **kwargs):
  45.         conn, cursor = self.get_conn_cursor()
  47.         cursor.execute(sql, kwargs)
  48.         result = cursor.fetchone()
  50.         self.close_conn_cursor(conn, cursor)
  51.         return result
  53.     def fetch_all(self, sql, **kwargs):
  54.         conn, cursor = self.get_conn_cursor()
  56.         cursor.execute(sql, kwargs)
  57.         result = cursor.fetchall()
  59.         self.close_conn_cursor(conn, cursor)
  61.         return result
  64. db = DBHelper()
  1. from db import db
  3. db.exec("insert into d1(name) values(%(name)s)", name="武沛齐666")
  5. ret = db.fetch_one("select * from d1")
  6. print(ret)
  8. ret = db.fetch_one("select * from d1 where id=%(nid)s", nid=3)
  9. print(ret)
  11. ret = db.fetch_all("select * from d1")
  12. print(ret)
  14. ret = db.fetch_all("select * from d1 where id>%(nid)s", nid=2)
  15. print(ret)

上下文管理

如果你想要让他也支持 with 上下文管理。
好处：多个SQL共用同一个连接，进一步提升性能。

  1. with 获取连接：
  2.     执行SQL（执行完毕后，自动将连接交还给连接池）
  1. # db_context.py
  2. import threading
  3. import pymysql
  4. from dbutils.pooled_db import PooledDB
  6. POOL = PooledDB(
  7.     creator=pymysql,  # 使用链接数据库的模块
  8.     maxconnections=5,  # 连接池允许的最大连接数，0和None表示不限制连接数
  9.     mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
  10.     maxcached=3,  # 链接池中最多闲置的链接，0和None不限制
  11.     blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
  12.     setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
  13.     ping=0,
  14.     host='127.0.0.1',
  15.     port=3306,
  16.     user='root',
  17.     password='root',
  18.     database='db1',
  19.     charset='utf8'
  20. )
  23. class Connect(object):
  24.     def __init__(self):
  25.         self.conn = conn = POOL.connection()
  26.         self.cursor = conn.cursor(pymysql.cursors.DictCursor)
  28.     def __enter__(self):
  29.         return self
  31.     def __exit__(self, exc_type, exc_val, exc_tb):
  32.         self.cursor.close()
  33.         self.conn.close()
  35.     def exec(self, sql, **kwargs):
  36.         self.cursor.execute(sql, kwargs)
  37.         self.conn.commit()
  39.     def fetch_one(self, sql, **kwargs):
  40.         self.cursor.execute(sql, kwargs)
  41.         result = self.cursor.fetchone()
  42.         return result
  44.     def fetch_all(self, sql, **kwargs):
  45.         self.cursor.execute(sql, kwargs)
  46.         result = self.cursor.fetchall()
  47.         return result
  1. from db_context import Connect
  3. with Connect() as obj:
  4.     # print(obj.conn)
  5.     # print(obj.cursor)
  6.     ret = obj.fetch_one("select * from d1")
  7.     print(ret)
  9.     ret = obj.fetch_one("select * from d1 where id=%(id)s", id=3)
  10.     print(ret)