Python数据库封装实现代码示例解析

在Python中操作数据库，我们可以使用第三方库或者原生的库实现，但是在实际开发过程中，我们经常会使用封装好的数据库操作库来简化操作。下面，我们将介绍如何封装数据库操作库，并给出实例说明。

数据库封装的好处

使用封装好的数据库操作库，可以简化开发者的操作，提高效率，减少重复的代码编写和出错的可能性。此外，通过合适的封装，操作代码的可读性和可维护性也会显著提高。

封装步骤

封装数据库操作库需要以下几个步骤：

1. 建立连接
2. 创建并执行sql语句
3. 关闭连接

接下来，我们通过一个MySQL数据库操作库的实现，来演示这些步骤。

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语句

在获得连接后，就可以编写具体的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()

示例

示例1：插入一条用户数据

下面我们用上述封装好的 insert_user 函数来插入一条用户数据。

data = {
    'name': 'zhangsan',
    'age': 23,
    'sex': 'male',
    'province': 'Gansu',
    'city': 'Lanzhou'
}

insert_user(data)

如果插入成功，则表 user 中将增加一条数据。

示例2：查询

下面代码展示如何使用一个函数，查询用户身份证号码，加入正确则返回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数据库操作库的一个简单实现，但它展示了封装原子操作模块的思路，并能在快捷、贴近生产环境的示例中帮你学会如何把这些思路应用到实际中。当你获得自己的经验，并精简出属于自己的方法，你的代码自然就会变得更加高效和具有可维护性。