简化async.run和Aio Web框架的AioMySQL的封装

2024-02-07 02:13:34

AioMySQL 是一个用纯 Python 编写的、快速、简单、完全异步的 MySQL 客户端。它基于 asyncio 实现，使它非常适合用于构建高性能、可扩展的网络应用程序。

async.run 是一个 Python 3.7 中引入的新函数，它允许您轻松地运行异步函数。它将自动创建并启动一个事件循环，并将异步函数的返回值作为结果返回。

Aio Web框架是一个基于 asyncio 的 Web 框架，它允许您轻松地构建高性能、可扩展的 Web 应用程序。

封装 AioMySQL

我们可以通过创建一个自定义类来封装 AioMySQL，该类将提供一个简单、统一的接口来访问 MySQL 数据库。

import asyncio
import aiomysql

class AioMySQLPool:
    def __init__(self, host, port, user, password, database):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.database = database
        self.pool = None

    async def connect(self):
        self.pool = await aiomysql.create_pool(
            host=self.host,
            port=self.port,
            user=self.user,
            password=self.password,
            db=self.database,
            autocommit=True
        )

    async def close(self):
        if self.pool:
            self.pool.close()
            await self.pool.wait_closed()

    async def execute(self, query, args=None):
        async with self.pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute(query, args)
                return await cur.fetchall()

使用封装后的 AioMySQL

现在，我们可以使用封装后的 AioMySQL 来简化在 async.run 和 Aio Web 框架中的使用。

async.run

import asyncio

async def main():
    pool = AioMySQLPool('localhost', 3306, 'root', 'password', 'database')
    await pool.connect()

    results = await pool.execute('SELECT * FROM users')
    for row in results:
        print(row)

    await pool.close()

asyncio.run(main())

Aio Web框架

import aiohttp_web
import aiomysql

app = aiohttp_web.Application()

async def handler(request):
    pool = request.app['pool']

    results = await pool.execute('SELECT * FROM users')
    return aiohttp_web.json_response(results)

app.add_routes([aiohttp_web.get('/users', handler)])

if __name__ == '__main__':
    pool = AioMySQLPool('localhost', 3306, 'root', 'password', 'database')
    await pool.connect()

    app['pool'] = pool
    aiohttp_web.run_app(app)

    await pool.close()