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

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

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

SQLAlchemy的使用

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
❯ 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里面

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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实现更多的方法:

1
2
3
4
5
6
7
8
9
10
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感受一下整体的用法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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

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

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

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

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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个字段:

  1. mp_id,它对应mps这个表里面对应记录的id字段
  2. user_id,它对应users这个表里面对应记录的id字段

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

1
2
3
4
5
6
7
8
9
10
11
12
13
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的方式,理由是:

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

使用Walrus我也创建了基类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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的写法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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的。