1. Python 的@contextmanager装饰器
我们使用sqlalchemy
的ORM
机制进行数据保存的时候默认是开启事务的。只有我们在commit
之后数据才真正保存到数据库。但是,当commit
出现故障就会导致数据导入失败,这时我们需要回滚事务。
1 try:
2 gift = Gift()
3 gift.uid = current_user.id
4 db.session.add(gift)
5 db.session.commit()
6 except Exception as e:
7 db.session.rollback()
8 raise e
但是,我们每次都是这样使用try..except
异常捕捉,很不高效。。
我们知道使用上下文协议可以写出一个上下文管理器,除此之外我们也可以使用装饰器来实现。
1 from contextlib import contextmanager
2
3
4
5 def ():
6 print('《', end='')
7 yield
8 print('》', end='')
9
10
11 with book_mark():
12 print('红烧肉', end='')
13
14
15 《红烧肉》
上面就是使用装饰器写了一个上下文管理器,yield
不一定需要返回什么,yield
的返回就相当于__enter__
的返回,作为as
的指向。执行顺序是先执行yield
之前的代码,然后再执行之后的代码。
contextmanager
可以将一个不是上下文管理器的类(或方法)变成一个上下文管理器。
我们简化上面一开始的代码,但是db.session.commit()
是第三方类库的方法,我们可以通过继承第三方类来实现新增方法。
1 from flask_sqlalchemy import SQLAlchemy as _SQLAlchemy
2
3 # 这里一个小技巧 就是讲第三方类起个别名
4
5
6 class SQLAlchemy(_SQLAlchemy):
7 @contextmanager
8 def auto_commit(self, throw=True):
9 try:
10 yield
11 self.session.commit()
12 except Exception as e:
13 self.session.rollback()
14 current_app.logger.exception('%r' % e)
15 if throw:
16 raise e
17
18 # 这里实例化的是我们继承过后的自己实现的类
19 db = SQLAlchemy()
20
21 # 使用
22 with db.auto_commit():
23 gift = Gift()
24 gift.uid = current_user.id
25 db.session.add(gift)
我们看到yield
之前也是可以为空的,执行的顺序是先执行with
下方的代码,然后之前yield
之前的代码,最后执行yield
之后的代码。
学习一个复杂的语法,我们先编写一个简单的例子,然后再使用复杂的用法
2. 创建时间的默认值
网上查了下创建时间的默认值问题,这里我是讲的是类的实例变量和类变量的区别,其实是调用问题。
我们先看下老师的思想
class Base(db.Model):
__abstract__ = True
create_time = Column('create_time', Integer)
status = Column(SmallInteger, default=1)
def __init__(self):
self.create_time = int(datetime.now().timestamp())
这里把时间赋值放在了实例方法中,在每次实例化类的时候都会自动赋值。
其实我们是可以使用默认值的
update_time = db.Column(db.DateTime, default=datetime.now,onupdate=datetime.now)
这里我们输入的是方法datetime.now
而不是方法的返回值datetime.now()
。
当我们输入的是返回值的时候,所有的时间都会是项目启动的那一刻,这就导致了实例化的时候,获取到的类变量是一个已经赋值了的,就不是一个实时的。
模板渲染是最消耗服务器性能的。
3. 重写 filter_by 函数
为什么要重写filter_by
方法呢?
当我们在所有的查询中都要包含同一个条件的时候,这时候就可以重写filter_by
来简化工作。
如何重写呢?
我们先看下继承关系
from flask_sqlalchemy import BaseQuery
class BaseQuery(orm.Query):
"""The default query object used for models, and exposed as
:attr:`~SQLAlchemy.Query`. This can be subclassed and
replaced for individual models by setting the :attr:`~Model.query_class`
attribute. This is a subclass of a standard SQLAlchemy
:class:`~sqlalchemy.orm.query.Query` class and has all the methods of a
standard query as well.
"""
class Query(object):
"""ORM-level SQL construction object.
:class:`.Query` is the source of all SELECT statements generated by the
ORM, both those formulated by end-user query operations as well as by
high level internal operations such as related collection loading. It
features a generative interface whereby successive calls return a new
:class:`.Query` object, a copy of the former with additional
criteria and options associated with it.
:class:`.Query` objects are normally initially generated using the
:meth:`~.Session.query` method of :class:`.Session`, and in
less common cases by instantiating the :class:`.Query` directly and
associating with a :class:`.Session` using the :meth:`.Query.with_session`
method.
For a full walkthrough of :class:`.Query` usage, see the
:ref:`ormtutorial_toplevel`.
"""
def filter_by(self, **kwargs):
r"""apply the given filtering criterion to a copy
of this :class:`.Query`, using keyword expressions.
e.g.::
session.query(MyClass).filter_by(name = 'some name')
Multiple criteria may be specified as comma separated; the effect
is that they will be joined together using the :func:`.and_`
function::
session.query(MyClass).
filter_by(name = 'some name', id = 5)
The keyword expressions are extracted from the primary
entity of the query, or the last entity that was the
target of a call to :meth:`.Query.join`.
.. seealso::
:meth:`.Query.filter` - filter on SQL expressions.
"""
clauses = [_entity_descriptor(self._joinpoint_zero(), key) == value
for key, value in kwargs.items()]
return self.filter(sql.and_(*clauses))
我们使用的flask_sqlalchemy
导入查询类BaseQuery
继承的最终是sqlalchemy
的Query
。
我们看到filter_by
是存在于sqlalchemy
中的,不过我们重写的时候,只要继承flask_sqlalchemy
的BaseQuery
就好了。
class Query(BaseQuery):
def filter_by(self, **kwargs):
if 'status' not in kwargs.keys():
kwargs['status'] = 1
# 完成原有的,通过调用基类方法
return super(Query, self).filter_by(**kwargs)
上面我们就为每个查询增加一个状态为 1 的判断。
重写好的查询类如何使用呢?
我们只需要在实例化SQLAlchemy
的时候指定查询类就好了
db = SQLAlchemy(query_class=Query)
思想:当我们不能直接修改源码的时候,我们可以通过继承,重写父类方法来完成我们的需求
链式调用:链式调用我们暂时理解为没有结束的查询。有一个主体 Query,不确定个查询过滤函数,最终遇到first
或者all
等函数才算结束链式调用。个人觉得很有用,针对一个链式调用,我们可以加上不同的查询判断分开成很多个链式调用。
4. 业务逻辑的编写方案
之前学习的时候老师也探讨过这个问题,建议是将可以抽象出来的,能够公用的业务逻辑尽量写在模型类中。
例如获取最近赠送礼物的一段业务逻辑(以老师鱼书代码为例)代码,老师的建议是放在模型类中。
@classmethod
def recent(cls):
gift_list = cls.query.filter_by(launched=False).order_by(
desc(Gift.create_time)).group_by(Gift.book_id).limit(
current_app.config['RECENT_BOOK_PER_PAGE']).all()
return gift_list
这里我大致总结一下老师的思想:
对于业务逻辑,一般建议编写在视图函数和模型对象中(以类方法,实例方法,静态方法的形式)。对于不可抽象的业务逻辑一般写在视图函数中,对于可抽象的业务逻辑一般写在模型对象中。
良好的封装是优秀代码的基础。
对于函数返回,我们尽量返回有提示的结果,例如字典返回。
5. 自定义JSONEncoder
from flask.json import JSONEncoder as _JSONEncoder
class JSONEncoder(_JSONEncoder):
def default(self, o):
if hasattr(o, 'keys') and hasattr(o, '__getitem__'):
return dict(o)
# 序列化时间
if isinstance(o, date):
return o.strftime('%Y-%m-%d')
raise ServerError()
class Flask(_Flask):
json_encoder = JSONEncoder
这样在使用jsonify的时候,就可以将MyObject类型的实例转化到json格式。
问题思考:
- 如果项目中有很多自定义Class,都有这个需求,JSONEncoder的实现似乎会很丑陋,有什么好的方式?
- 同一个Class在不同的场景,json后要保留的字段差别如果很大,该如果实现?
对该问题的实现,可以采用以下方式:
#在模型对象中重写dict方法
class User(Base):
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(50), unique=True, nullable=False)
nickname = db.Column(db.String(24), unique=True)
auth = db.Column(db.SmallInteger, default=1) # 1是普通用户,2是管理员
_password = db.Column('password', db.String(100))
#自定义将模型对象转成dict ,即dict(user)
#dict(user),会先调用keys方法,这里重写,自定义获取返回的字段
def keys(self):
return ['id', 'email', 'nickname', 'auth']
# dict(user)获取完字段后,会取出对应字段的值,这里使用__getitem__,这里getattr(self, item)拿到值信息,item为key名
def __getitem__(self, item):
return getattr(self, item)
以上的keys和getitem方法,会在自定义的jsonencoder方法中dict(对象)时候调用,如下:
#自定义json方法,在方法中实现对象转字典返回,这需要在模型对象中实现对应dict方法
class JSONEncoder(_JSONEncoder):
def default(self, o):
return dict(o)
调用方法如:
def get_user(uid):
user = User.query.get_or_404(uid)
return jsonify(user)
所以这样就实现了jsonencoder将对象转化成字典,然后直接jsonnfy调用序列化。并且复用性强,在模型中定义dict调用的方法,使得转化的字典key可以控制。