Python基础教程

第一章: 环境搭建,安装Python

第二章: 挑选一款趁手的IDE

第三章: 计算机基础知识

第四章: 命令行基础知识

第五章: 从全局把握Python

第六章: Python语言基础

第七章: Python流程控制

第八章: Python数据类型与运算

第九章: Python字符串类型

第十章: Python列表类型

第十一章: Python元组类型

第十二章: Python字典类型

第十三章: Python集合类型

第十四章: Python函数处理

第十五章: Python文件处理

第十六章: Python面向对象

第十七章: Python异常处理

第十八章: Python模块处理

第十九章: Python高级编程

第二十章: Python项目实战

首页 > Python基础教程 > 第十八章: Python模块处理 > 18.6节: 程序实战-封装数据库模块

18.6节: 程序实战-封装数据库模块

薯条老师 2022-12-08 15:04:18 22888 0

编辑 收藏

广州番禺Python, Java小班周末班培训

薯条老师在广州做Python和Java的小班培训,一个班最多10人,学员的平均就业薪资有11K。不在广州的同学可以报名线上直播班,跟线下小班的同学们同步学习。培训的课程有Python爬虫,Python后端开发,Python办公自动化,Python大数据分析,Python量化投资,Python机器学习,Java中高级后端开发。授课详情请点击:http://chipscoco.com/?cate=6

18.6.1 操作数据库

一个完整的Python后端项目,其架构通常会包含一个数据访问层,利用该层提供的接口,就可以对数据库进行读写操作。在本节程序实战中,我们使用单例模式来封装一个用于操作MySQL的Python库,通过该库,即可方便地对数据库进行操作。关于MySQL的基础知识,同学们可以学习薯条老师写的这套MySQL基础教程:http://chipscoco.com/?cate=13

18.6.2 安装PyMySQL

MySQL基于C-S架构,这里的C是客户端,S是服务端。在程序中连接MySQL服务器,需通过MySQL提供的客户端接口来进行操作。而PyMySQL正是实现了MySQL客户端接口的一个库。安装PyMySQL可直接使用PIP大法,很简单:

pip install pymysql

18.6.3 PyMySQl用法简介

通过以下步骤来使用PyMySQL来操作MySQL:

(1) 通过pymysql的connect方法构造一个数据库连接对象

(2) 执行数据库连接对象的cursor方法来构造一个游标对象

(3) 在游标对象中执行execute或executemany方法对mysql数据库进行增查改删以及管理,对于插入,删除,更新等操作,需要通过commit方法进行事务提交。

(4) 在游标对象中执行fetchone或fetchall方法来获取在execute或executemany方法中执行的结果

pymysql的connect方法:

pymysql.connect(host="localhost", user, passwd, db, charset="utf8")

参数解释:

host表示mysql服务器的地址,默认值为localhost, 表示本机,user为数据库的账号,passwd为数据库的密码,db为特定的数据库, charset为数据库的字符编码,chrset必需指定正确的数据库编码,否则会出现乱码。

游标对象的execute方法:

cursor.execute(query, args=None)

query表示数据库的查询语句,查询语句中可带参数,用%s来表示。args:序列类型,query中带%s参数时,必须使用args来传递实参值。假设查询商品表中id为1的记录,SQL为"select id, name, price from goods where id=%s"则args为:(1,)。execute方法的返回值是一个整型值,表示影响的行数。

游标对象的executemany方法:

cursor.executemany(query, args)

query表示数据库的查询语句,查询语句中可带参数,用%s来表示。args:序列类型,query中带%s参数时,必须使用args来传递实参值。通常使用executemany来执行数据的批量读写,假设在employee表中进行数据的批量写,SQL为"insert into employee(name) values(%s)",此时可以将多个实参值组合到一个列表中,args为[("小旭",),("小董",)]。executemany方法的返回值是一个整型值,表示影响的行数。

游标对象的fetchone方法:

cursor.fetchone()

fetchone的返回值类型为一个元组,表示一个单行的记录,值为None时,表示没有数据。

游标对象的fetchall方法:

cursor.fetchall()

fetchall方法的返回值类型为一个元组,表示多行的记录值,值为None时,表示没有数据。

18.6.4 编写一个配置模块

为方便项目的维护,我们先编写一个配置模块,将用户的登录,密码等消息都写入配置文件,这样当配置消息发生变动时,就无需修改源代码。模块名为settings.py, 模块中的代码如下所示:

class DatabaseType:
    def __init__(self):
        pass
    MYSQL = 0
    ORACLE = 1
    REDIS = 2
    
    
DBConfig_OFFICIAL = {
    DatabaseType.MYSQL: {
        "host": "", # 配置线上环境的MySQL服务器的地址
        "port": 3306, # 配置线上环境的MySQL服务器的端口号
        "user": "", # 配置线上环境的MySQL服务器的用户名
        "password": "", # 配置线上环境的MySQL服务器的登录密码
        "database": "" # 配置要连接的线上环境的数据库
    },
}


DBConfig_TEST = {
    DatabaseType.MYSQL: {
        "host": "", # 配置测试环境的MySQL服务器的地址
        "port": 3306, # 配置测试环境的MySQL服务器的端口号
        "user": "",# 配置测试环境的MySQL服务器的用户名
        "password": "", # 配置测试环境的MySQL服务器的登录密码
        "database": "" # 配置要连接的测试环境的数据库
    },
}


DBConfig = DBConfig_OFFICIAL

18.6.5 用单例模式来进行封装

关于单例模式,可参考薯条老师写的这篇教程:http://chipscoco.com/?id=357。现在我们使用单例模式来封装一个对MySQL进行读写操作的Python模块database.py。模块的代码如下所示:


# __author__ = 薯条老师

import pymysql
from config.settings import DBConfig, DatabaseType


class Database:
    """
    对数据库操作进行了简单的封装
    """
    # 类属性__db__instances是一个字典类型,用来保存数据库的实例
    __db_instances = {}
    
    
    @classmethod
    def get_instance(cls, db_type=DatabaseType.MYSQL):
        """
        定义get_instance类方法,用来获取数据库对象的单例
        所谓的单例就是一个类只有一个实例,调用该方法每次获取到
        的都是同一个数据库实例,,type默认为MYSQL类型,表示
        默认获取的是mysql的数据库实例
        """
        if db_type not in cls.__db_instances:
            # 如果不存在,就构造一个Database的实例对象
            cls.__db_instances[db_type] = Database(db_type)
        return cls.__db_instances[db_type]
        
        
    def __init__(self, db_type=DatabaseType.MYSQL):
        """
        :param db_type: 数据库的类型,数据库的类型在DatabaseType中进行了定义
        默认为MYSQL类型,表示创建mysql类型的数据库实例
        """
        self.__db_type = db_type
        self.__db = self.__get_database()
        self.__cursors = {}
        
        
    def __get_database(self):
        db = None
        # 根据类型字段,来创建对应的数据库实例
        if self.__db_type == DatabaseType.MYSQL:
            try:
                db = pymysql.connect(
                                     host=DBConfig[DatabaseType.MYSQL]["host"],
                                     user=DBConfig[DatabaseType.MYSQL]["user"],
                                     password=DBConfig[DatabaseType.MYSQL]["password"],
                                     database=DBConfig[DatabaseType.MYSQL]["database"],
                                     port=DBConfig[DatabaseType.MYSQL]["port"]
                                     )
            except IOError:
                db = None
        return db
        
        
    def batch_insert(self, sql=None, args=None):
        """
        :param sql: 客户端传递的查询语句
        :param args: 查询语句对应的参数
        :param data: 批量插入的数据
        :return:True表示批量写入成功,False表示失败
        """
        status, last_rowid, error = True, -1, ""
        if not self.__db:
            return status
        if self.__db_type == DatabaseType.MYSQL:
            # 如果数据库的实例对象为MySQLdb,则执行executemany方法来进行批量写入
            if "mysql" not in self.__cursors:
                self.__cursors["mysql"] = self.__db.cursor()
            try:
                self.__cursors["mysql"].executemany(sql, args)
                last_rowid = self.__db.insert_id()
                self.__db.commit()
                pass
            except (Exception, ) as e:
                status = False
                error = e
        return status, last_rowid, error
        
        
    def create_database(self, **params):
        """
        :param params: 可变参数,
        params中的name表示数据库名,body表示创建数据库的额外参数
        :return: 返回一个状态信息,True表示创建成功,False表示创建失败
        """
        status = True
        if self.__db_type == DatabaseType.ELASTICSEARCH:
            es_index = params.get("name", None)
            mappings = params.get("body", None)
            if not self.__db.indices.exists(index =params["name"]):
                try:
                    self.__db.index(index=es_index, body=mappings)
                except(Exception, ):
                    status = False
        return status
        
        
    def query(self, ql, *args):
        """
        :param ql:表示查询语句
        :param args:表示查询的参数
        :return:
        """
        data = None
        if self.__db_type == DatabaseType.MYSQL:
            if "mysql" not in self.__cursors:
                self.__cursors["mysql"] = self.__db.cursor()
            if not args:
                self.__cursors["mysql"].execute(ql)
            else:
                self.__cursors["mysql"].execute(ql, args)
            data = self.__cursors["mysql"].fetchall()
        return data

18.6.6 数据库模块的使用

模块封装好以后,就可以将其作为数据访问层提供给其它模块进行使用。现在我们将配置模块settings.py保存在config目录中,将数据库封装模块database.py保存至dao中。这里的config作为项目的配置层,dao作为项目的数据访问层:

image.png

app.py表示项目的入口文件,我们现在在入口文件中写一个简单的测试程序, 代码如下所示:

"""
@author: 薯条老师
@desc: 对数据库模块进行测试
"""
from config.settings import DatabaseType
from dao.database import  Database


if __name__ == "__main__":
    mysql_conn = Database.get_instance(db_type=DatabaseType.MYSQL)
    sql = "select id, name from test where id=%s"
    id_ = 1
    rows = mysql_conn.query(sql, id_)
    print(rows)

18.6.7 最具实力的小班培训

薯条老师在广州做Python和Java的小班培训,一个班最多10人。不在广州的同学可以报名线上直播班,跟线下小班的同学们同步学习打算参加小班培训的同学,必须遵守薯条老师的学习安排,认真做作业和项目。把知识学好,学扎实,那么找到一份高薪的工作就是很简单的一件事。

(1) Python后端工程师高薪就业班,月薪11K-18K,免费领取课程大纲
(2) Python爬虫工程师高薪就业班,年薪十五万,免费领取课程大纲
(3) Java后端开发工程师高薪就业班,月薪11K-20K, 免费领取课程大纲
(4) Python大数据分析,量化投资就业班,月薪12K-25K,免费领取课程大纲

扫码免费领取Python学习资料:


欢迎 发表评论: