广西建设厅招投标中心网站,wordpress 小说 采集,页面设置自定义wordpress,游戏推广员是做什么的操作数据库封装SQL工具类的两种方式
为了更方便的实现基于连接池和pymysql 连接数据库#xff0c;需开发一个sql工具类来让sql操作更简洁用两张方式来封装SQL工具类
1 #xff09;单例模式
封装 db.py 工具类
import pymysql
from dbutils.pooled_db import PooledDBclas…操作数据库封装SQL工具类的两种方式
为了更方便的实现基于连接池和pymysql 连接数据库需开发一个sql工具类来让sql操作更简洁用两张方式来封装SQL工具类
1 单例模式
封装 db.py 工具类
import pymysql
from dbutils.pooled_db import PooledDBclass DBHelper(object):def __init__(self):self.pool PooledDB(creatorpymysql,maxconnections5,mincached2,maxcached3,blockingTrue,setsession[],ping0,host127.0.0.1,port3306userroot,passwordxxxxx,databaseuserdb,charsetutf8)def get_conn_cursor(self):conn self.pool.connection()cursorconn.cursor(pyymsql.cursors.DictCursor)return conn, cursordef 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 resultdef 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 dbv1 db.fetch_one(select * from d1)
print(v1)v2 db.fetch_one(select * from d1 where id%(nid)s , nid3)
print(v2)2 ) 上下文管理
基于 with 上下文管理
with 获取连接:执行sql (执行完毕后自动将连接交还给连接池)封装 db_context.py
import threading
import pymysql
from dbutils.pooled_db import PooledDBPOOL PooledDB(creatorpymysql, # 使用连接数据库的模块maxconnections5,mincached2,maxcached3,blockingTrue,setssion[],ping0,host127.0.0.1port3306,userroot,passwordxxxx,databaseuserdb,charsetutf8
)class Connect(object):def __init__(self):self.conn conn POOL.connection() # 连接self.cursor conn.cursor(pymysql.cursors.DictCursor) # 游标def __enter__(self):return selfdef __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 resultdef fetch_all(self, sql, **kwargs):self.cursor.excute(sql, kwargs)result self.cursor.fetchall()return resultyyy.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, id3)print(ret)
