Python3编程实战Tetris机器人（数据库操作）

Posted 2021-09-04 zhoutk

系列文章入口

《Python3编程实战Tetris机器人》

设计思路

将用户手动玩和AI自动玩游戏的历史记录下来，存入数据库，供后面进行分析。为了不依赖某个特定的数据系统，设计了一个通用数据库操作接口，以方便在应用层面切换不同的数据库。

接口设计

class BaseDao(object):
    def select(self, tablename, params={}, fields=None):     # 查询接口，参数：数据表名；查询参数（ORM规则融入字典中，请参看下一篇日志）；返回数据字段
        fields = [] if fields == None else fields            # Python的默认参数行为很是不同，会记录上一次调用的结果，有点象其它语言中的静态变量
        return dbhelper.select(tablename, params, fields)    # 真正的查询实现，dbhelper是针对特定数据库的接口实现

    def insert(self, tablename, params={}, fields=[]):       # 新增接口，CURD函数的参数形式一致，这种设计方便作rest微服务时，由http的不同访问方式直接选择后台操作方法
        if \'_id_\' in params and len(params) < 2 or \'_id_\' not in params and len(params) < 1:     # 要求提供_id_，约定_id_为所有表的主键
            return {"code": 301, "err": "The params is error."}
        return dbhelper.insert(tablename, params)

    def update(self, tablename, params={}, fields=[]):
        if \'_id_\' not in params or len(params) < 2:
            return {"code": 301, "err": "The params is error."}
        return dbhelper.update(tablename, params)

    def delete(self, tablename, params={}, fields=[]):
        if \'_id_\' not in params:
            return {"code": 301, "err": "The params is error."}
        return dbhelper.delete(tablename, params)

    def querySql(self, sql, values = [], params = {}, fields = []):  # 手写查询接口
        return dbhelper.querySql(sql, values, params, fields)

    def execSql(self, sql, values = []):      # 手写非查询接口
        return dbhelper.exec_sql(sql, values)

    def insertBatch(self, tablename, elements : List):     # 批量写入接口
        return dbhelper.insertBatch(tablename,elements)

    def transGo(elements = [], isAsync = False):    # 事务接口，待实现
        pass

具体实现（Sqlit3）

首先实现了对Sqlit3的操作接口。

直接面对Sqlit3的函数

def exec_sql(sql, values, opType = 0): # opType : 0 - 单条SQL语句； 1 - 批量操作语句；2 - 查询返回数据集；
    try:    # 所有与数据操作都通过这个函数，需要用异常处理封装
        flag = False    # 是否出错标识变量
        error = {}
        if not os.path.exists("./dist"):  # 存储位置目录存在判断
            os.mkdir("dist")
        conn = dbHandle.connect("./dist/log.db")  # 连接数据库或新建
        cur = conn.cursor()
        if opType == 1:          # 批量操作
            num = cur.executemany(sql, values)
        else:                    # 单条语句
            num = cur.execute(sql, values)
        if opType == 2:          # 有结果集返回
            result = cur.fetchall()
        else:
            conn.commit()
        # print(\'Sql: \', sql, \' Values: \', values)
    except Exception as err:     # 出错
        flag = True
        error = err
        print(\'Error: \', err)
    finally:
        conn.close()            # 结束处理，并格式化返回结果
        if flag:
            return False, error, num if \'num\' in dir() else 0
    return True, result if \'result\' in dir() else [], len(result) if opType == 2 else num.rowcount if \'num\' in dir() else 0

查询函数

这里讲解函数主干，关于在字典中融入ORM的解析，请参看下篇日志。

def select(tablename, params={}, fields=None, sql = None, values = None):
    where = ""
    AndJoinStr = \' and \'
    reserveKeys = {}
    for rk in ["sort", "search", "page", "size", "sum", "count", "group"]:
        # 提取保留关键字
    for k, v in params.items():
        whereExtra = ""
        if k == "ins":
            # 保留关键字ins,lks,ors处理
        else:
            flag = False
            if type(v) == "str":
                # 不等查询操作处理
            elif reserveKeys.get(\'search\'):
                # 精确查询与模糊查询处理
            else:
                whereExtra += k + " =? "
                values.append(v)
        where += whereExtra
    # 排序、统计、分组和分页等操作处理
    rs = exec_sql(sql, values, 2)
    return {"code": 200, "rows": rs[1], "total": rs[2]}

插入函数

删除和更新与插入类似，这里讲解插入函数

def insert(tablename, params={}):
    sql = "insert into %s ( " % tablename     # 主干
    ks = params.keys()
    vs = []
    ps = ""
    for al in ks:           # 解析参数
        sql += al + ","     # 按插入语句拼接每一个参数
        ps += "?,"          # python的sqlit3封装没法送入list对象来实现元组数据的写入，只能拆开
        vs.append(params[al])
    sql = sql[:-1] + ") values (" + ps[:-1] + ")"  # 去掉最后的逗号，并完成sql语句
    rs = exec_sql(sql, vs)                         # 执行，vs参数中不能嵌套list，比起C++版本的实现，这里有些别扭
    if rs[0]:                                      # 返回结果
        return {"code": 200, "info": "create success.", "total": rs[2]}
    else:
        return {"code": 701, "error": rs[1].args[0], "total": rs[2]}

批量插入

事实证明，一条条的插入效率太低，AI运行时，数据写入跟不上节奏。

def insertBatch(tablename, elements : List):
    if len(elements) == 0:    # 无输入元素，直接退出
        return {"code": 201, "info": "There is no elements exist.", "total": 0}
    elif len(elements) == 1:  # 只有一个元素，调用insert实现来完成
        return insert(tablename, elements[0])

    sql = "insert into %s ( " % tablename
    isFirst = True      # 在循环的第一次，要处理操作字段
    vs = []
    ps = ""
    for ele in elements:
        if isFirst:
            isFirst = False
            ks = ele.keys()
            for al in ks:   # 操作字段，只需处理一次
                sql += al + ","
                ps += "?,"  # 参数批占位符
        items = []
        for bl in ks:  # 按key的顺序逐个添加写入参数值，字典的访问顺序是不一定的
            items.append(ele[bl])
        vs.append(items)
    sql = sql[:-1] + ") values (" + ps[:-1] + ")"  # 最后的拼接
    rs = exec_sql(sql, vs, 1)    # 执行
    if rs[0]:                    # 返回结果
        return {"code": 200, "info": "create success.", "total": rs[2]}
    else:
        return {"code": 701, "error": rs[1].args[0], "total": rs[2]}

内容预告

下一篇日志讲解融入字典中的ORM规则设计及使用方法，有了这一套规则，无需要再写sql语句。欲后事如何，请持续关注，谢谢！

项目地址

https://gitee.com/zhoutk/ptetris
或
https://github.com/zhoutk/ptetris

运行方法

1. install python3, git
2. git clone https://gitee.com/zhoutk/ptetris (or download and unzip source code)
3. cd ptetris
4. python3 tetris

This project surpport windows, linux, macOs

on linux, you must install tkinter first, use this command:  
sudo apt install python3-tk

相关项目

已经实现了C++版，项目地址：

https://gitee.com/zhoutk/qtetris