参悟python元类(又称metaclass)系列实战(四)

  • Z时代
  • 2024-01-10
  • 分类：综合

参悟python元类(又称metaclass)系列实战(四)[Python基础]

写在前面

在上一章节参悟python元类(又称metaclass)系列实战(三)完成了users类和users表的字段映射;

在继续丰富users类的操作之前, 我们花一章来完成mysql的链接;

有误的地方恳请大神指正下。

创建全局连接池, 以求避免频繁地打开和关闭数据库连接

  • 因为查询耗时, 计划引入异步, 即async/await(不懂没关系, 就当是普通代码, 只是调用方式特别而已)

    $ pip3 install aiomysql    # 通过pip安装

  • 创建Mysql类, 定义静态方法createPool

    import aiomysql
    class Mysql:
        @staticmethod
        async def createPool():
            """连接池由全局变量__pool存储，缺省情况下将编码设置为utf8，自动提交事务"""
            try:
                global __pool
                __pool = await aiomysql.create_pool(
                    host="localhost",        # mysql服务器地址
                    port=3306,               # mysql监听端口
                    user="root",             # mysql用户名
                    password="passwd",       # mysql密码
                    db="ormdemo",            # mysql实例名(database)
                    charset="utf8",          # 编码设置
                    autocommit=True,         # 自动提交insert/update/delete
                    maxsize=10,              # 连接池上限
                    minsize=1                # 连接池下限
                )
                print("创建成功")
            except Exception as e:
                print(f"连接 mysql 出错:{e}")
    # 测试代码
    if __name__ == "__main__":
        import asyncio
        loop = asyncio.get_event_loop()
        loop.run_until_complete(Mysql.createPool())

  • 给Mysql类新增静态方法select、execute

    @staticmethod
    async def select(sql, args, size=None):
        """传入SQL语句和SQL参数
        @sql :  sql语句
        @args:  条件值列表
        @size:  查询的结果集大小
        """
        print(f"{sql} ==> {args}")
        # 使用with上下文管理器, 自动执行close
        async with __pool.acquire() as conn:
            # acquire: 从空闲池获得连接。如果需要，并且池的大小小于maxsize，则创建新连接。
            async with conn.cursor(aiomysql.DictCursor) as cur:
                # cursor: 连接的游标
                # DictCursor: 作为字典返回结果的游标
                await cur.execute(sql.replace("?", "%s"), args or ())
                # sql.replace("?", "%s"): "SELECT * FROM users WHERE uid=?".replace("?", "%s")
                # args or (): "SELECT * FROM users WHERE uid=%s" % (101,)
                if size:
                    rs = await cur.fetchmany(size)
                else:
                    rs = await cur.fetchall()
            print(f"rows returned: {len(rs)}")
            return rs
    @staticmethod
    async def execute(sql, args, autocommit=True):
        """
        INSERT、UPDATE、DELETE操作，可以通用的execute函数，
        因为这3种SQL的执行都需要相同的参数, 返回一个整数表示影响的行数
        """
        print(f"{sql} ==> {args}")
        async with __pool.acquire() as conn:
            if not autocommit:
                await conn.begin()           # 开始处理
            try:
                async with conn.cursor(aiomysql.DictCursor) as cur:
                    await cur.execute(sql.replace("?", "%s"), args)
                    affected = cur.rowcount  # 受影响的行数
                if not autocommit:
                    await conn.commit()      # 提交更改, 非自动提交时才需要
            except Exception as e:
                if not autocommit:
                    await conn.rollback()    # 回滚更改
                raise Exception(e)
            return affected

测试下select

if __name__ == "__main__":
    import asyncio
    loop = asyncio.get_event_loop()
    loop.run_until_complete(Mysql.createPool())
    rs = loop.run_until_complete(Mysql.select("select uid, name from users where uid=?", [101]))
    print(rs)

  • 控制台输出

创建成功
select * from users where uid=? ==> [101]
rows returned: 1
[{"uid": 101, "name": "z417"}]

总结

  1. 定义了Mysql类, 其中有3个静态方法

  2. 引入了异步, 调用方式共分两步

以上是 参悟python元类(又称metaclass)系列实战(四) 的全部内容， 来源链接： utcz.com/z/537852.html

回到顶部