Flask之七：sqlite3数据库（原生SQL）

基本思路

• 写一个数据库操作的实例db，这个实例绑定具体的sqlite数据库，并且可以对这个数据库执行SQL语句
• 各级蓝图都导入这个实例，通过这个实例直接用SQL语句操作数据库
• 但是：由于现在ORM比较方便，而且ORM也可以通过SQL语句来直接操作数据库，所以这个实例的意义不大

注意：各级文件的路径是从哪里开始算起的

数据库操作实例db

# dbconn.py
# --------------------------------
"""
    使用说明:

    使用时只需要指定sqlite3数据库的路径, 然后把这个文件中的db导入, 然户就可以使用db来操作它绑定的数据库了

    Sqlite3数据库管理员db，负责执行各种SQL语句；
    用法：导入db, db.exe()、db.exep(,)、db.sql()
    返回: （成功与否T/F，查询到的行数，附加信息，查询结果数据集=含元组的list）

    1. Sqlite3Admin类是操作sqlite3数据库的类
    2. 方法==exe==用于：执行insert|delete|update语句
        接受参数灵活：
            1. 一条sql语句的str
            2. 一组sql语句str组成的list
        返回：（成功与否T/F，受影响的行数，附加信息）
    3. 方法==exep==用于执行带参数的sql语句，主要是insert
        接受参数
            1-sql语句模版，如： insert into user values (null, ?, ?)
            2-参数的列表，如 (('Jim', 27), ('Tim', 56), ('Janny', 21))，也很灵活, list和tuple混用也行, 但是建议全部用list：
                [[], []]、([], [])、((), ())、[(), ()]、[[], ()]
        返回：（成功与否T/F，受影响的行数，附加信息）
    4. 方法==sql==用于执行select语句，返回查询结果
        接受参数：
            sql_str_list: SQL语句，str/list,如果是list则只取其中第一个字符串
            返回: （成功与否T/F，查询到的行数，附加信息，查询结果数据集=含元组的list）
    """


# 我自己的py模块, 主要是用SQL语句原生操作数据库的, 不依赖其他的模块
import sqlite3


# 唯一需要设计的内容: sqlite3数据库路径
sqlite3_database_file_path_from_root = 'sqlite3/my.db'  # 文件路径要从主py执行起算，因为这些模块最终都是导入到主py文件里去执行的，都是以主py文件所在的地方（root根目录）作为查找文件的起点


class Sqlite3Basic(object):
    """
    Sqlite3数据库连接的基础类，负责数据库连接、断开；用于后续子类的扩展
    """
    sqlite3_name = sqlite3_database_file_path_from_root  # 数据库路径及文件名
    obj_conn = None  # 数据库连接对象
    obj_cursor = None  # 游标对象

    # 连接数据库，并创建游标
    @classmethod
    def conn(cls):
        cls.obj_conn = sqlite3.connect(cls.sqlite3_name)  # 创建数据库链接
        cls.obj_cursor = cls.obj_conn.cursor()  # 创建游标

    # 关闭游标，并断开数据库
    @classmethod
    def disconn(cls):
        cls.obj_cursor.close()  # 关闭游标
        cls.obj_conn.commit()  # 提交事务
        cls.obj_conn.close()  # 关闭数据库链接


class Sqlite3Admin(Sqlite3Basic):
    # 执行sql语句进行insert delete update操作，返回受影响的行数
    @classmethod
    def exe(cls, sql_str_list):
        """
        执行一条/多条 insert|update|delete 语句(exe=执行)

        千万注意！insert时不要忘记对字符串加''！
        :param sql_str_list: SQL语句，可以是str（一句）也可以是list[str]（多句）
        :return: （成功与否T/F，受影响的行数，附加信息）
        """
        # 1. 连接数据库
        cls.conn()
        # 2. 执行操作语句
        rowcount = 0  # 记录受影响的条数
        # 如果sql语句是一个list，那就一条一条的执行
        if isinstance(sql_str_list, list):  # 推荐用isinstance代替type这里
            for sql_i in sql_str_list:
                cls.obj_cursor.execute(sql_i)
                rowcount += cls.obj_cursor.rowcount
        # 不然就只有一句sql语句
        else:
            cls.obj_cursor.execute(sql_str_list)
            rowcount = cls.obj_cursor.rowcount
        # 3. 关闭数据库
        cls.disconn()
        # 4. 设限返回值
        if rowcount == -1:  # -1表示发生错误
            return False, rowcount, '错误：未知错误！请检查！'
        elif rowcount == 0:
            return False, rowcount, '错误：受影响的数据条数=[0]，请检查！'
        else:
            return True, rowcount, '成功'

    # 同时执行多条sql语句 executemany
    @classmethod
    def exep(cls, sql_model: str, para_list_tuple):
        """
        执行多条sql(一般是insert)语句，效率更高，exep=执行带参数（para）

        千万注意！insert时不要忘记对字符串加''！
        :param sql_model: sql模版语句，如 'insert into user values (null, ? ?)'
        :param para_list_tuple: ?参数的list，本身及成员可以是list或者tuple，最后都要转为tuple
        :return: （成功与否T/F，受影响的行数，附加信息）
        """
        # 1. 连接数据库
        cls.conn()
        # 2. 执行操作语句
        # 如果成员是list，则转为tuple
        para_list_tuple = [tuple(p) for p in para_list_tuple if isinstance(p, list)]
        # 如果参数本身是list，则转为tuple
        para_list_tuple = tuple(para_list_tuple) if not isinstance(para_list_tuple, tuple) else para_list_tuple
        # 执行executemany()方法
        cls.obj_cursor.executemany(sql_model, para_list_tuple)
        rowcount = cls.obj_cursor.rowcount  # 记录受影响的条数
        # 3. 关闭数据库
        cls.disconn()
        # 4. 设限返回值
        if rowcount == -1:  # -1表示发生错误
            return False, rowcount, '错误：未知错误！请检查！'
        elif rowcount == 0:
            return False, rowcount, '受影：响的数据条数=[0]，请检查！'
        else:
            return True, rowcount, '成功'

    # 查询语句，返回数据集
    @classmethod
    def sql(cls, sql_str_list):
        """
        执行 select 语句

        :param sql_str_list: SQL语句，str/list,如果是list则只取其中第一个字符串
        :return: （成功与否T/F，查询到的行数，附加信息，查询结果数据集=含元组的list）
        """
        # 1. 连接数据库
        cls.conn()
        # 2. 执行查询语句
        # 兼容一下：如果参数的一个list，则取其第一个作为sql查询语句
        sql_str_list = sql_str_list[0] if isinstance(sql_str_list, list) else sql_str_list
        # 执行
        cls.obj_cursor.execute(sql_str_list)
        rs = cls.obj_cursor.fetchall()  # 这是一个list，里面是tuple: [(), ()]
        rowcount = len(rs)
        # 3. 关闭数据库
        cls.disconn()
        # 4. 返回值
        if rowcount == -1:  # -1表示发生错误
            return False, rowcount, '错误：未知错误！请检查！', []
        else:
            return True, rowcount, '成功，查询结果[%d]条' % rowcount, rs


# 数据库实例，直接使用db的方法就可以了！其他py文件直接导入去用
db = Sqlite3Admin()


# 测试
if __name__ == '__main__':
    # 关于不停的left join、left join的用法
    # 写作：TABLE1 left   outer        join TABLE2 on    TABLE1.A = TABLE2.C
    # 读作： 表1   在左边 不匹配也不删行 关联   表2  条件是      这个==这个
    sql = ('select e.id, e.name, e.age, b.name, ea.address from employee as e '
           'left outer join bumen as b on e.bumen_id = b.id '
           'left outer join employee_address as ea on e.id = ea.employee_id')
    print(db.sql(sql))