wechat-admin:ORM使用篇
/ / / 阅读数:4540我是典型的「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 的。