在Python中操作数据库,我们可以使用第三方库或者原生的库实现,但是在实际开发过程中,我们经常会使用封装好的数据库操作库来简化操作。下面,我们将介绍如何封装数据库操作库,并给出实例说明。
使用封装好的数据库操作库,可以简化开发者的操作,提高效率,减少重复的代码编写和出错的可能性。此外,通过合适的封装,操作代码的可读性和可维护性也会显著提高。
封装数据库操作库需要以下几个步骤:
接下来,我们通过一个MySQL数据库操作库的实现,来演示这些步骤。
在使用之前,需要安装 pymysql
库。
pip install pymysql
首先需要定义 MySQL 配置信息,以便拥有访问MySQL数据库的权限。可以创建一个 config.py
文件进行配置信息的存储:
MYSQL_HOST = 'localhost'
MYSQL_PORT = 3306
MYSQL_USER = 'root'
MYSQL_PASSWORD = 'password'
MYSQL_DB = 'testdb'
在使用的时候,需要建立连接,但是连接数不能太多,因此需要使用连接池来进行管理。我们可以使用 pool
模块,将连接创建封装成 init_connection
函数。
import pymysql
from pymysql import connections, cursors
from DBUtils.PooledDB import PooledDB
from config import *
class MysqlPool(object):
def __init__(self):
self.pool = None
self._connect_pool()
def _connect_pool(self):
self.pool = PooledDB(
creator=pymysql,
maxconnections=10,
mincached=2,
maxcached=5,
maxshared=3,
blocking=True,
host=MYSQL_HOST,
port=MYSQL_PORT,
user=MYSQL_USER,
password=MYSQL_PASSWORD,
database=MYSQL_DB,
cursorclass=cursors.DictCursor
)
def get_connection(self):
return self.pool.connection()
pool = MysqlPool()
通过以上代码,我们建立了一个名为 MysqlPool
的MySQL连接池,其中的 _connect_pool
函数使用对应的配置信息连接到MySQL数据库,并使用 PooledDB
对它进行封装。get_connection
函数用于获取连接池的连接。
在获得连接后,就可以编写具体的SQL语句了。以下是一个插入用户数据的示例,他将用户数据写入到表 user
中。
def insert_user(user_data):
conn = pool.get_connection()
sql = """
insert into user (name, age, sex, province, city)
values (%s, %s, %s, %s, %s)
"""
data = (user_data['name'], user_data['age'], user_data['sex'],
user_data['province'], user_data['city'])
try:
with conn.cursor() as cursor:
cursor.execute(sql, data)
conn.commit()
except Exception as e:
conn.rollback()
raise e
finally:
conn.close()
以上代码,在函数内建立连接,将用户传入的信息用 SQL 语句进行插入,并进行异常处理(例如数据插入失败,数据回滚)。
在执行完 SQL 后需要对连接进行关闭,使用以下代码:
conn.close()
下面我们用上述封装好的 insert_user
函数来插入一条用户数据。
data = {
'name': 'zhangsan',
'age': 23,
'sex': 'male',
'province': 'Gansu',
'city': 'Lanzhou'
}
insert_user(data)
如果插入成功,则表 user
中将增加一条数据。
下面代码展示如何使用一个函数,查询用户身份证号码,加入正确则返回True:
def check_id_card(id_card):
conn = pool.get_connection()
sql = """
SELECT count(*) as cnt FROM user WHERE id_card=%s
"""
data = (id_card,)
try:
with conn.cursor() as cursor:
cursor.execute(sql, data)
result = cursor.fetchone()
return result['cnt'] > 0
except Exception as e:
raise e
finally:
conn.close()
assert check_id_card('411423199809210078') == False
以上代码使用sql查询的方法,获取该身份证在用户表中的计数,并将计数结果的Bool值作为函数返回值,用以判定身份证号码是否唯一。
本文仅是MySQL数据库操作库的一个简单实现,但它展示了封装原子操作模块的思路,并能在快捷、贴近生产环境的示例中帮你学会如何把这些思路应用到实际中。当你获得自己的经验,并精简出属于自己的方法,你的代码自然就会变得更加高效和具有可维护性。
本文链接:http://task.lmcjl.com/news/14910.html