Sanic 学习(三) - 开发一个数据库扩展

今天继续以实操方式来学习 Python 语言下最快的 Web 开发框架之一：Sanic。在之前的两篇实践文章中，我已经完成了 Session 和访问限流功能扩展，这篇打算来一个在 Web 开发中同样必不可少的扩展功能：数据库操作。

因为只是体验开发思路和流程，所以我选择了简便易用的 SQLite 数据库。跟之前一样，先来看看我想要实现的功能应用代码：

app = Sanic('zzxworld')

@app.get('/')
async def home(request):
    # 获取用户列表
    users = await User.all()

    # 拼装每个用户的 HTML 代码
    userHtmlItems = map(lambda u: '<tr>'
                       '<td>'+str(u['id'])+'</td>'
                       '<td>'+u['name']+'</td>'
                       '</tr>', users)

    # 拼装用户列表区块的 HTML 代码
    userBlockHtml = '''<table border="1">
        <thead><tr><th>ID</th><th>Name</th></tr></thead>
        %s
        </table>''' % ''.join(userHtmlItems)

    return html(userBlockHtml)

可以看出，在上面的代码中，users = await User.all() 这句是关键。我需要实现一个用户表的模型类：User，其中需要有一个获取所有用户的 all 方法。那就先创建一个这样的 Python 类：

class DBModel:
    """数据库表基础模型"""

    @property
    def db(self):
        """获取数据库连接对象"""
        return DBConnection().getConnection()

    async def execute(self, *args, **kwargs):
        """执行 SQL 语句并返回查询对象"""
        return await self.db.execute(*args, **kwargs)

class User(DBModel):
    """用户数据表模型"""

    @classmethod
    async def all(cls):
        """获取所有用户"""
        cursor = await cls().execute('select * from users')
        items = await cursor.fetchall()
        await cursor.close()

        return items

    @classmethod
    async def init(cls):
        """初始化用户数据"""
        # 创建用户表
        cursor = await cls().execute('CREATE TABLE users('
                         'id INTEGER PRIMARY KEY,'
                         'name VARCHAR(30))')
        await cursor.close()

        # 插入用户示例数据
        cursor = await cls().execute('INSERT INTO users (name) VALUES'
                         '("Tom"),'
                         '("Jack"),'
                         '("zzxworld")')
        await cursor.close()

跟之前一样，关键部分我都加上了注释，所以具体的功能逻辑我就不赘述了，来讲讲相关 Python 知识点和实现目的。

上面的代码涉及到了 Python 语言的三个概念，不熟悉 Python 的朋友需要先去了解一下。一是类属性装饰器 @property，二是类方法装饰器 @classmethod，还有一个是异步协程语法：async 和 await。
我创建了一个 DBModel 类，拿来用作数据表模型的基础类，其中可以写一些数据库操作的公共方法。
User 类继承 DBModel 类，然后添加了两个类方法。all 用来查询并返回所有用户数据，init 用来创建 users 表并插入一些演示数据。

上面的代码实现了我在开头的 Sanic 应用代码中想要达成的目的，不过也挖了一个新的坑需要填，也就是 DBConnection().getConnection() 这行获取数据库连接的代码。继续来完成这部分代码：

import aiosqlite

class DBConnection:
    """数据库连接对象"""

    _instance = None
    _connection = None

    def __new__(cls, *args):
        """实现连接对象的单例模型"""
        if cls._instance is None:
            cls._instance = super(DBConnection, cls).__new__(cls)
        return cls._instance

    async def connection(self):
        """创建并获取数据库连接"""
        if self._connection is None:
            self._connection = await aiosqlite.connect(':memory:')
            self._connection.row_factory = aiosqlite.Row
        return self._connection

    def getConnection(self):
        """获取数据库连接"""
        return self._connection

    async def disconnection(self):
        """关闭数据库连接"""
        if self._connection is not None:
            await self._connection.close()

上面的代码使用了一个支持异步协程的 SQLite 包：aiosqlite，并用 Python 的 __new__ 方法实现了单例模式。这个连接对象提供了数据库的连接，断开和获取链接的方法。因为只是做测试，所以 SQLite 的连接路径为内存：:memory:，这样每次程序重启都会是一个全新的数据库环境。

至此，我想要在 Sanic 中实现的数据库功能基本已经达成。最后再补充一个 Sanic 的扩展，实现程序启动前自动连接数据库，并初始化用户数据表；程序终止前自动断开数据库的功能。

from sanic_ext import Extension, Extend

class Database(Extension):
    """zzxworld 的数据库扩展"""
    name = 'zzxDatabase'

    def startup(self, bootstrap):
        """扩展入口"""
        self.app.before_server_start(self.connection)
        self.app.before_server_stop(self.deconnection)

    @staticmethod
    async def connection(app):
        """创建数据库连接"""
        db = DBConnection()
        app.ctx.db = await db.connection()
        await User.init()

    @staticmethod
    async def deconnection(app):
        """关闭数据库连接"""
        if hasattr(app.ctx, 'db'):
            await app.ctx.db.close()

Extend.register(Database)

前面已经完成了数据库操作的核心代码，在扩展编写这里就省事了很多。调用写好的方法并注册到 Sanic 框架的处理流程中就可以了。我这个小巧的 Sanic 数据库扩展到这里也算是大功告成。

Sanic 学习(三) - 开发一个数据库扩展

上一篇

下一篇