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方法](https://github.com/dongweiming/wechat-
admin/blob/master/ext.py#L8)：


❯ 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里面](https://github.com/dongweiming/wechat-
admin/blob/master/models/mixin.py#L4)：


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我也创建了[基类](https://github.com/dongweiming/wechat-
admin/blob/master/models/redis.py#L1)：


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的[写法](https://github.com/dongweiming/wechat-
admin/blob/master/models/setting.py)：


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的。

版权声明：本文由董伟明原创，未经作者授权禁止任何微信公众号和向掘金(juejin.im)转载，技术博客转载采用保留署名-非商业性使用-禁止演绎 4.0-国际许可协议
python

SQLAlchemy的使用

Walrus的使用

结语