wechat-admin：ORM使用篇

我是典型的「ORM 党」。ORM 全称 Object Relational Mapping，中文叫作对象关系映射。通过它我们可以直接使用 Python 的类的方式做数据库开发，不用直接写原生的 SQL 语句（甚至不需要 SQL 的基础），使用 ORM 有如下优点：

易用性。使用这种 ORM 数据库抽象封装方式做开发可以有效减少重复 SQL 语句出现的概率，写出来的模型也更直观、清晰。
设计灵活。可以很轻松地写复杂的查询。

另外提一下，我在工作中其实有一半时间还是需要直接写 SQL 的，不过用类的方式包装起来用了。可能不太好理解，有兴趣的可以看一下豆瓣开源的 douban-orz 这个项目，很多场景都是使用这种数据管理方案，我觉得还是蛮好用的。

SQLAlchemy 的使用

SQLAlchemy 是业界最流行的 ORM 库，它支持多个关系数据库引擎，如 MySQL、PostgreSQL 等数据库，可以近乎无痛地换数据库。本项目的联系人、群聊、公众号等关系和数据都存在了 MySQL 上。当使用一个 ORM 库，基于业务特点和开发者个人习惯通常都会定义一些基类或者 Mixin 类，我写的项目大都会添加 to_dict 方法：

❯ cat ext.py
from datetime import datetime

from sqlalchemy import Column, DateTime
from flask_sqlalchemy import SQLAlchemy, Model


class BaseModel(Model):
    # create_at这个属性也是创建表结构默认都包含的
    create_at = Column(DateTime, default=datetime.utcnow())

    def to_dict(self):
        columns = self.__table__.columns.keys()
        return {key: getattr(self, key) for key in columns}


db = SQLAlchemy(model_class=BaseModel)

凡是后端 API 用于返回数据的都需要把一个对象中需要的属性和值拼成一个 json 对象。

我是直接在创建 db 时就把 to_dict 和 create_at「注入」进去了，不过这样的方法不能使用 db 这个属性，对于数据库操作的就不方便这么用了。我另外有个习惯是添加 create 方法，方法内创建对象然后提交事务，相当于封装一个方法完成创建 / 返回以后的实例，这个我放在了 Mixin 里面：

from ext import db


class BaseMixin(object):
    @classmethod
    def create(cls, **kw):
        session = db.session
        if 'id' in kw:
            obj = session.query(cls).get(kw['id'])
            if obj:
                return obj
        obj = cls(**kw)
        session.add(obj)
        session.commit()
        return obj

另外还会继承这个 BaseMixin 实现更多的方法：

class CoreMixin(BaseMixin):

    @property
    def avatar(self):
        return avatar_tmpl.format(self.id)

    def to_dict(self):
        rs = super().to_dict()
        rs['avatar'] = self.avatar
        return rs

avatar 这个属性是用户 / 群聊 / 公众号类需要的，但是 Message 类不需要，所以独立的实现。我们拿 Group 感受一下整体的用法：

from sqlalchemy.ext.hybrid import hybrid_method, hybrid_property

from ext import db
from .mixin import BaseMixin


class Group(CoreMixin, db.Model):
    __tablename__ = 'groups'
    __table_args__ = {'mysql_charset': 'utf8mb4'}
    id = db.Column(db.String(20), primary_key=True)  # puid
    owner_id = db.Column(db.String(20), index=True)
    nick_name = db.Column(db.String(60), index=True)

    def __repr__(self):
        return '<Group %r>' % self.nick_name

    @hybrid_method
    def is_member(self, user):
        return user in self.members

    @hybrid_property
    def count(self):
        return len(self.members)

    def to_dict(self):
        rs = super().to_dict()
        rs['count'] = self.count
        return rs

为了演示，我省略了一些业务用到的方法。解释下一下：

hybrid_method 和 hybrid_property 是 SQLAlchemy 提供的混合机制，使用它们可以给一个 db.Model 类添加额外的方法或者属性。
to_dict 方法已经被重载多次了，每次通过 super ().to_dict () 拿到原来的结果然后添加新的内容。
加了__table_args__是因为可能会有一些 utf8 字符集未包含的内容，需要扩大这个字符集。

不过事情远没有这么简单，因为选择 MySQL 这个关系型数据库，就是由于项目需求是有「关系」的：

用户和联系人。比如 A 的联系人 B 和 A 互相关注，但是 A 中的群聊有个成员 C，A 和 C 并没有关注关系。
用户和群聊。用户和对应的群聊也是有关系的，我们需要了解 A 是不是群聊 B 内的成员
用户和公众号。用户和公众号也是有关系的，我们需要了解 A 有没有关注公众号 B，而在结构上公众号和用户很像。

要实现这样的关系，需要先定义三张表来存放这个关系：

friendship = db.Table(
    'friends',
    db.Column('user_id', db.String(20), db.ForeignKey('users.id')),
    db.Column('friend_id', db.String(20), db.ForeignKey('users.id')),
    mysql_charset='utf8mb4'
)


group_relationship = db.Table(
    'group_relationship',
    db.Column('group_id', db.String(20), db.ForeignKey('groups.id'),
              nullable=False),
    db.Column('user_id', db.String(20), db.ForeignKey('users.id'),
              nullable=False),
    mysql_charset='utf8mb4'
)


mp_relationship = db.Table(
    'mp_relationship',
    db.Column('mp_id', db.String(20), db.ForeignKey('mps.id'),
              nullable=False),
    db.Column('user_id', db.String(20), db.ForeignKey('users.id'),
              nullable=False),
    mysql_charset='utf8mb4'
)

举个例子更好明白，mp_relationship 包含 2 个字段：

mp_id，它对应 mps 这个表里面对应记录的 id 字段
user_id，它对应 users 这个表里面对应记录的 id 字段

铺垫完成了，感受下 User 类如何定义关系的：

class User(CoreMixin, db.Model):
    __tablename__ = 'users'
    ...
    groups = db.relationship('Group', secondary=group_relationship,
                             backref='members')
    mps = db.relationship('MP', secondary=mp_relationship,
                          backref='users')
    friends = db.relationship('User',
                              secondary=friendship,
                              primaryjoin=(friendship.c.user_id == id),
                              secondaryjoin = (friendship.c.friend_id == id),
                              lazy = 'dynamic'
                              )

groups 和 mps 的用法很像，定义字段的时候使用 db.relationship，其中 secondary 参数就是上面的关系表对象。backref 表示在对应的类（Group 或者 MP）中的属性名字。

friends 要更复杂，因为 friendship 中的 2 个字段都在同一张表，所以有 2 个外键，可以使用 primaryjoin 明确联结条件，secondaryjoin 来指定多对多关系中的二级联结条件。lazy 决定了 SQLAlchemy 什么时候从数据库中加载数据，dynamic 表示只是返回一个查询对象而不是直接加载这些数据，这样在加载数据前我们可以在执行语句中添加过滤之类的条件。

Walrus 的使用

Walrus 是 Redis 的 ORM 库，和 SQLAlchemy 相比名气差了很多，我觉得大家不怎么用 ORM 操作 Redis 的主要原因是 Redis 就是个内存数据库，它的使用不像 SQL 那样容易写错，最多就是使用 pipeline，过程很清晰，操作和查询都很简单。

限于公司技术栈，我其实在工作中也很少用 Redis，需求很简单就直接调用对应方法了。这次是我想尝试一下 ORM 的方式，理由是：

手写操作和查询还是会出现语句重复利用率不高的问题
就像前面说的，更喜欢通过操作 ORM 的开发方式

使用 Walrus 我也创建了基类：

from walrus import Database, Model, ListField, SetField, HashField
from config import REDIS_URL

db = Database.from_url(REDIS_URL)
LISTENER_TASK_KEY = 'listener:task_id'


class RBase(Model):
    __database__ = db

    def to_dict(self):
        data = {}
        for name, field in self._fields.items():
            if name in self._data:
                val = self._data[name]
                data[name] = val if field._as_json else field.db_value(val)
            else:
                if isinstance(field, ListField):
                    type_func = list
                elif isinstance(field, SetField):
                    type_func = set
                elif isinstance(field, HashField):
                    type_func = dict
                else:
                    type_func = lambda x: x
                data[name] = type_func(getattr(self, name))
        return data

    @classmethod
    def get(cls, id):
        try:
            return super().get(cls.id == id)
        except ValueError:
            return cls.create(id=id)

同样的实现了 to_dict 方法。不过我重写了 get 方法：get 不到就创建，这是由于业务需要，第一次拿不到就要创建一个默认配置的记录来用。看一下 model 的写法：

from walrus import TextField, ListField, JSONField, IntegerField
from .redis import RBase
from config import welcome_text, invite_text, kick_text, group_patterns


class GroupSettings(RBase):
    id = TextField(primary_key=True)
    welcome_text = TextField(default=welcome_text)
    invite_text = TextField(default=invite_text)
    group_patterns = JSONField(default=group_patterns)
    creators = ListField()
    mp_forward = JSONField(default=[])
    kick_quorum_n = IntegerField(default=5)
    kick_period = IntegerField(default=5)
    kick_text = TextField(default=kick_text)

可以看到 GroupSettings 都是带默认值的，所以创建的时候传入最重要的 id 就可以。

结语

在 wechat-admin 中就是这么用 ORM 的。