MySQL中的开发基于Python的SQL工具类操作数据库简单示例

操作数据库封装SQL工具类的两种方式

• 为了更方便的实现基于连接池和pymysql 连接数据库，需开发一个sql工具类来让sql操作更简洁
• 用两张方式来封装SQL工具类

1 ）单例模式

封装 db.py 工具类

import pymysql
from dbutils.pooled_db import PooledDB

class DBHelper(object):
  def __init__(self):
      self.pool = PooledDB(
        creator=pymysql,
        maxconnections=5,
        mincached=2,
        maxcached=3,
        blocking=True,
        setsession=[],
        ping=0,
        host='127.0.0.1',
        port=3306
        user='root',
        password='xxxxx',
        database='userdb',
        charset='utf8'
      )
  
  def get_conn_cursor(self):
    conn = self.pool.connection()
    cursor=conn.cursor(pyymsql.cursors.DictCursor)
    return conn, cursor
  
  def close_conn_cursor(self, *args):
    for item in args:
      item.close()
  
  def exec(self, sql, **kwargs):
    conn, cursor = self.get_conn_cursor()
    cursor.execute(sql, kwargs)
    conn.commit()
    self.close_conn_cursor(conn, cursor)

  def fetch_one(self, sql, **kwargs):
    conn, cursor = self.get_conn_cursor()
    cursor.execute(sql, kwargs)
    result = cursor.fetchone()
    self.cloes_conn_cursor(conn, cursor)
    return result
  
  def fetch_all(self, sql, **kwarrgs):
    conn, cursor = self.get_conn_cursor()
    cursor.execute(sql, kwargs)
    result = cursor.fetchall()
    self.close_conn_cursor(conn, cursor)
  
db = DBHelper()

xxx.py 调用示例

from db import db

v1 = db.fetch_one("select * from d1")
print(v1)

v2 = db.fetch_one('select * from d1 where id=%(nid)s ', nid=3)
print(v2)

2 ) 上下文管理

基于 with 上下文管理

with 获取连接:
  执行sql (执行完毕后，自动将连接交还给连接池)

封装 db_context.py

import threading
import pymysql
from dbutils.pooled_db import PooledDB

POOL = PooledDB(
  creator=pymysql, # 使用连接数据库的模块
  maxconnections=5,
  mincached=2,
  maxcached=3,
  blocking=True,
  setssion=[],
  ping=0,
  host='127.0.0.1'
  port=3306,
  user='root',
  password='xxxx',
  database='userdb',
  charset='utf8'
)

class Connect(object):
  def __init__(self):
      self.conn = conn = POOL.connection() # 连接
      self.cursor = conn.cursor(pymysql.cursors.DictCursor) # 游标

  def __enter__(self):
      return self
  
  def __exit__(self, exc_type, exc_val, exc_tb):
      self.cursor.close()
      self.conn.close()

  def exec(self, sql, **kwargs):
      self.cursor.execute(sql, kwargs)
      self.conn.commit()

  def fetch_one(self, sql, **kwargs):
      self.cursor.execute(sql, kwargs)
      result = self.cursor.fetchone()
      return result
  
  def fetch_all(self, sql, **kwargs):
      self.cursor.excute(sql, kwargs)
      result = self.cursor.fetchall()
      return result

yyy.py 调用示例

from db_context import Connect

### 实例化 对象得到值
with Connect() as obj:
  ret = obj.fetch_one('select * from d1')
  print(ret)

  ret = obj.fetch_one("select * from d1 where id=%(id)s", id=3)
  print(ret)