当前位置：首页 > news >正文

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

news 2025/8/14 14:58:47

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

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

1 ）单例模式

封装 db.py 工具类

import pymysql
from dbutils.pooled_db import PooledDBclass 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=3306user='root',password='xxxxx',database='userdb',charset='utf8')def get_conn_cursor(self):conn = self.pool.connection()cursor=conn.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 ', nid=3)
print(v2)

2 ) 上下文管理

基于 with 上下文管理

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

封装 db_context.py

import threading
import pymysql
from dbutils.pooled_db import PooledDBPOOL = 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 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 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)