专题章节目录

慕课网Flask高级编程实战-知识点思维导图

慕课网Flask高级编程实战-1.项目准备 和 Flask入门

慕课网Flask高级编程实战-2.搜索书籍路由编写

慕课网Flask高级编程实战-3.蓝图、模型与CodeFirst

慕课网Flask高级编程实战-4.flask核心机制

慕课网Flask高级编程实战-5.书籍详情页面的构建

慕课网Flask高级编程实战-6.书籍详情页面的构建

慕课网Flask高级编程实战-7.静态文件、模板、消息闪现与Jinja2

慕课网Flask高级编程实战-8.用户登录与注册

慕课网Flask高级编程实战-9.书籍交易模型（数据库事务、重写Flask中的对象）

慕课网Flask高级编程实战-10.鱼书业务处理

慕课网Flask高级编程实战-11.Python与Flask的结合应用

慕课网Flask构建可扩展的RESTful API-1. 起步与红图

慕课网Flask构建可扩展的RESTful API-2. REST基本特征

慕课网Flask构建可扩展的RESTful API-3. 自定义异常对象

慕课网Flask构建可扩展的RESTful API-5. Token与HTTPBasic验证 —— 用令牌来管理用户

慕课网Flask构建可扩展的RESTful API-6. 模型对象的序列化

慕课网Flask构建可扩展的RESTful API-7. 权限控制

11.1 重置密码流程分析

重置密码主要流程如下：

image.png

其中，发送重置密码邮件后的流程如下：

image.png

---

11.2 first_or_404和可调用对象

1.first_or_404

视图函数接受用户填写的email账号，如果不存在应该跳转到404界面，这个逻辑flask-sqlalchemy为我们提供了良好的封装，不需要手动去处理,只需要调用Query的first_or_404()方法即可

user = User.query.filter_by(email=account_email).first_or_404()

first_or_404内部是对first函数的封装，他在获取了first的结果以后，会进行空判断，如果结果为空，则调用abort()方法，而abort()方法内部是调用了一个对象

first_or_404

def first_or_404(self):
"""Like :meth:`first` but aborts with 404 if not found instead of returning ``None``."""

rv = self.first()
if rv is None:
abort(404)
return rv

abort()

def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::

abort(404) # 404 Not Found
abort(Response('Hello World'))

Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::

abort(404)
abort(Response('Hello World'))

'''
return _aborter(status, *args, **kwargs)

_aborter = Aborter()

如果要将一个对象像函数一样调用，需要在函数内部实现__call__方法，_aborter(status, *args, **kwargs)实际上就是调用了Aborter的__call__方法

Aborter()

class Aborter(object):

"""
When passed a dict of code -> exception items it can be used as
callable that raises exceptions. If the first argument to the
callable is an integer it will be looked up in the mapping, if it's
a WSGI application it will be raised in a proxy exception.

The rest of the arguments are forwarded to the exception constructor.
"""

def __init__(self, mapping=None, extra=None):
if mapping is None:
mapping = default_exceptions
self.mapping = dict(mapping)
if extra is not None:
self.mapping.update(extra)

def __call__(self, code, *args, **kwargs):
if not args and not kwargs and not isinstance(code, integer_types):
raise HTTPException(response=code)
if code not in self.mapping:
raise LookupError('no exception for %r' % code)
raise self.mapping[code](*args, **kwargs)

2.可调用对象的作用

image.png

关于其中的统一调用接口，我们看下面的实例

class A:
def go(self):
return object()


class B:
def run(self):
return object()


def func():
return object()


def main(param):
# 我想在main中传入一个参数，得到一个对象object
# b.run()
# a.go()
# func()
pass


main(A())
main(B())
main(func)

如果不适用可调用对象，我们需要在main函数中区分是不同的情况，分别处理，非常的麻烦。
如果main方法传入的参数只有方法那就好说了，我们只需要param()就可以调用，所以python为我们提供了可调用对象，让对象可以像方法一样调用,修改好的代码如下

class A:
def __call__(self):
return object()


class B:
def __call__(self):
return object()


def func():
return object()


def main(callable):
# 我想在main中传入一个参数，得到一个对象object
callable()
pass


main(A())
main(B())
main(func)

这就是可调用对象中统一调用接口的意义。在flask中有很多可调用对象的使用，如globas.py中对current_app,request等对象的封装.其中_find_app，_lookup_req_object都是可调用对象

# context locals
_request_ctx_stack = LocalStack()
_app_ctx_stack = LocalStack()
current_app = LocalProxy(_find_app)
request = LocalProxy(partial(_lookup_req_object, 'request'))
session = LocalProxy(partial(_lookup_req_object, 'session'))
g = LocalProxy(partial(_lookup_app_object, 'g'))

---

11.3 HttpException

1.first_ot_404的异常抛出流程

我们来接着看Aborter类中的first_ot_404流程,在first_ot_404调用流程中，由于不满足__call__的前两个判断条件，最终会抛出self.mapping，其在构造函数中进行了声明

class Aborter(object):

"""
When passed a dict of code -> exception items it can be used as
callable that raises exceptions. If the first argument to the
callable is an integer it will be looked up in the mapping, if it's
a WSGI application it will be raised in a proxy exception.

The rest of the arguments are forwarded to the exception constructor.
"""

def __init__(self, mapping=None, extra=None):
if mapping is None:
mapping = default_exceptions
self.mapping = dict(mapping)
if extra is not None:
self.mapping.update(extra)

def __call__(self, code, *args, **kwargs):
if not args and not kwargs and not isinstance(code, integer_types):
raise HTTPException(response=code)
if code not in self.mapping:
raise LookupError('no exception for %r' % code)
raise self.mapping[code](*args, **kwargs)

self.mapping是一个dict，其封装了default_exceptions，下面来看一下default_exceptions的装载，在_find_exceptions中完成。他的作用是扫描当前模块下所有HTTPException的子类对象，并装载到default_exceptions

default_exceptions = {}
__all__ = ['HTTPException']


def _find_exceptions():
for name, obj in iteritems(globals()):
try:
is_http_exception = issubclass(obj, HTTPException)
except TypeError:
is_http_exception = False
if not is_http_exception or obj.code is None:
continue
__all__.append(obj.__name__)
old_obj = default_exceptions.get(obj.code, None)
if old_obj is not None and issubclass(obj, old_obj):
continue
default_exceptions[obj.code] = obj
_find_exceptions()
del _find_exceptions

最终Aborter函数的__call__方法拿着封装好的self.mapping（实质是default_exceptions）通过参数传来的code去匹配相应的异常，并进行抛出。

以下为first_ot_404的总执行流程

image.png

2.抛出异常后到浏览器异常界面的显示流程

因为first_or_404（）后面传入Aborter的code为404，所以其对应抛出的异常就是
NotFound对象,而其中的description描述文本，就是异常页面的显示文本

class NotFound(HTTPException):

"""*404* `Not Found`

Raise if a resource does not exist and never existed.
"""
code = 404
description = (
'The requested URL was not found on the server. '
'If you entered the URL manually please check your spelling and '
'try again.'
)

而所有的页面的显示文本，都是由response来做的，我们的试图函数在调用first_or_404()函数时，由于结果不存在，就抛出了上面的NotFound异常而终止了，后面的异常流程直到界面显示，都是由HttpException完成的，来看下其对应源码

# get_response最终返回了一个Response对象
def get_response(self, environ=None):
"""Get a response object. If one was passed to the exception
it's returned directly.

:param environ: the optional environ for the request. This
can be used to modify the response depending
on how the request looked like.
:return: a :class:`Response` object or a subclass thereof.
"""
if self.response is not None:
return self.response
if environ is not None:
environ = _get_environ(environ)
headers = self.get_headers(environ)
# get_body获取页面文本内容
return Response(self.get_body(environ), self.code, headers)

get_body()

def get_body(self, environ=None):
"""Get the HTML body."""
return text_type((
u'<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">\n'
u'<title>%(code)s %(name)s</title>\n'
u'<h1>%(name)s</h1>\n'
u'%(description)s\n'
) % {
'code': self.code,
'name': escape(self.name),
# self.get_description获取异常的description信息
'description': self.get_description(environ)
})

至此，first_or_404，从调用，到界面显示流程，就剖析完成了

3.基于AOP统一处理异常

如果我们不想返回默认的404界面，而想自己定制，那么直接想到的方法就是在代码中捕获异常，返回我们自定义的试图，但是这样代码太啰嗦了。我们可以采用AOP面向切面的设计思想，定义一个装饰器，用来统一处理某类异常，好在Flask已经为我们提供了这种方法，使用方法如下

# web是蓝图对象，当然也可以使用app对象
# app_errorhandler接受一个状态码，代表当前方法处理的异常
# not_found函数不是只能返回视图，他可以做任何你想做的事情
@web.app_errorhandler(404)
def not_found(e):
return render_template('404.html'), 404

---

11.4 发送电子邮件

我们使用flask_mail来完成电子邮件的发送

pipenv install flask-mail

1.在app中注册flask-mail

mail = Mail()
mail.init_app(mail)

2.EMAIL配置

# email配置
MAIL_SERVER = 'smtp.qq.com'
MAIL_PORT = 465
MAIL_USE_SSL = True
MAIL_USE_TSL = False
MAIL_USERNAME = '1152057576@qq.com'
# QQ邮箱->设置->账户->[POP3...]->生成授权码->发送短信->获取授权码
MAIL_PASSWORD = 'pstomjiomwyybadh'

3.编写邮件工具类

def send_email(to, subject, template, **kwargs):
msg = Message(
subject,
sender=current_app.config['MAIL_USERNAME'],
recipients=[to])
# 发送一封HTML邮件
mail.html = render_template(template, kwargs)
mail.send(msg)

4.测试调用

send_email(account_email, "重置你的密码", 'email/reset_password.html',
user=user, token='aaa')

5.使用itsdangerous生成token

我们的token应该有一个过期时间，应该可以存储我们想要存储的值，flask为我们提供了一个非常好用的插件itsdangerous

def generate_token(self, expiration=600):
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
s = Serializer(secret_key=current_app.config['SECRET_KEY'], expires_in=expiration)
# s.dumps生成的是byte数组，我们需要编码成字符串
return s.dumps({'id': self.id}).decode('utf-8')

6.重置密码

@classmethod
def reset_password(cls, token, new_password):
s = Serializer(secret_key=current_app.config['SECRET_KEY'])
try:
# 解析token获取用户id，方法与生成token相反
data = s.loads(token.encode('utf-8'))
except:
return False
uid = data.get(id)
with db.auto_commit():
# 获取用户信息并修改
user = User.query.get_or_404(uid)
user.password = new_password
return True

7.视图函数编写

@web.route('/reset/password/<token>', methods=['GET', 'POST'])
def forget_password(token):
form = ResetPasswordForm(request.form)
if request.method == 'POST' and form.validate():
success = User.reset_password(token, form.password1.data)
if success:
flash('您的密码已重置，请使用新密码登录')
return redirect(url_for('web.login'))

flash('密码重置失败')
return render_template('auth/forget_password.html')

8.发送邮件优化，异步发送

def send_mail_async(app, msg):
# App_Context 的栈Local Stack是线程隔离的，在新线程里栈顶为空，需要手动入栈
with app.app_context():
try:
mail.send(msg)
except:
print('邮件发送失败')


def send_mail(to, subject, template, **kwargs):
msg = Message(
'[鱼书]'+' '+subject,
sender=current_app.config['MAIL_USERNAME'],
recipients=[to])
msg.html = render_template(template, **kwargs)
# current_app是代理对象，在当前线程下有指向，但是在新开启的线程中就没了，因为LocalProxy是线程隔离的
app = current_app._get_current_object()
thr = Thread(target=send_mail_async, args=[app,msg])
thr.start()

---

11.5 鱼漂

1.业务逻辑编写

当用户在像其他用户发起一个赠书请求的时候，这实际上就是一次交易的发起，我们将这个交易起名为鱼漂（Drift），具体业务逻辑如下

image.png

2.模型设计

models/drift.py

class Drift(Base):
"""
一次具体的交易信息
"""
__tablename__ = 'drift'

id = Column(Integer, primary_key=True)

# 邮寄信息
recipient_name = Column(String(20), nullable=False)
address = Column(String(100), nullable=False)
message = Column(String(200))
mobile = Column(String(20), nullable=False)

# 书籍信息
isbn = Column(String(13))
book_title = Column(String(50))
book_author = Column(String(30))
book_img = Column(String(50))

# 请求者信息
requester_id = Column(Integer)
requester_nickname = Column(String(20))

# 赠送者信息
gifter_id = Column(Integer)
gift_id = Column(Integer)
gifter_nickname = Column(String(20))
# 状态
_pending = Column('pending', SmallInteger, default=1)

状态信息应该使用枚举类

class PendingStatus(Enum):
Waiting = 1
Success = 2
Reject = 3
Redraw = 4

模型冗余而不是模型关联？

1.Drift旨在记录历史状态，而模型关联记录的是实时关联的。所以应该在Drift中直接平铺所有信息

2.模型关联是使得每次查询的时候多次关联，降低查询速度

3.鱼漂条件检测

1.自己不能够向自己索要数据
models/gift.py

def is_yourself_gift(self, uid):
return uid == self.uid

2.鱼豆数量必须大于等于1
3.每索取两本书，必须赠送一本书
models/user.py

def can_send_drifts(self):
if self.beans < 1:
return False
success_gift_count = Gift.query.filter_by(
uid=self.id, launched=True).count()
success_receive_count = Drift.query.filter_by(
uid=self.id, pending=PendingStatus.Success).count()

return floor(success_receive_count / 2) <= success_gift_count

4.完成鱼漂业务逻辑

试图函数

web/drift.py

@web.route('/drift/<int:gid>', methods=['GET', 'POST'])
@login_required
def send_drift(gid):
current_gift = Gift.query.get_or_404(gid)
if current_gift.is_yourself_gift(current_user.id):
flash('这本书是自己的(*^▽^*)，不能向自己索要哦')
return redirect(url_for('web.book_detail', isbn=current_gift.isbn))

can = current_user.can_send_drifts()
if not can:
return render_template('not_enough_beans.html', beans=current_user.beans)

form = DriftForm(request.form)
if request.method == 'POST' and form.validate():
save_drift(form, current_gift)
send_mail(current_gift.user.email, '有人想要一本书', 'email/get_gift.html',
wisher=current_user,
gift=current_gift)
# 成功后跳转到鱼漂历史记录界面
return redirect(url_for('web.pending'))
# summary用户的简介频繁使用，且更像是一种用户的属性，所以作为用户的一个属性
gifter = current_gift.user.summary
return render_template('drift.html', gifter=gifter, user_beans=current_user.beans, form=form)

models/user.py

@property
def summary(self):
return dict(
nikename=self.nickname,
beans=self.beans,
email=self.email,
send_receive=str(self.send_counter) + '/' + str(self.receive_counter)
)

save_gift.py

def save_drift(drift_form, current_gift):
if current_user.beans < 1:
# TODO 自定义异常
raise Exception()

with db.auto_commit():
drift = Drift()
drift_form.populate_obj(drift)

drift.gift_id = current_gift.id
drift.requester_id = current_user.id
drift.requester_nickname = current_user.nickname
drift.gifter_nickname = current_gift.user.nickname
drift.gifter_id = current_gift.user.id

book = BookViewModel(current_gift.book)
drift.book_title = book.title
drift.book_author = book.author
drift.book_img = book.image
drift.isbn = book.isbn

db.session.add(drift)

current_user.beans -= 1

---

11.6 交易记录

1.业务逻辑

image.png

这两个条件应该是或者（or）的关系

2.获取鱼漂列表

# select * from drift where requester_id = ? or gifter_id = ?
# order by create_time desc
drifts = Drift.query.filter(
or_(Drift.requester_id == current_user.id,
Drift.gifter_id == current_user.id))\
.order_by(desc(Drift.create_time)).all()

3.Drift ViewModel 编写

Drift ViewModel需要适应当前用户是赠送者和当前用户是索要者两个情况

image.png

class DriftCollection:

def __init__(self, drifts, current_user_id):
self.data = []

self.data = self._parse(drifts, current_user_id)

def _parse(self, drifts, current_user_id):
return [DriftViewModel(drift, current_user_id).data for drift in drifts]


class DriftViewModel:

def __init__(self, drift, current_user_id):
self.data = {}

self.data = self._parse(drift, current_user_id)

@staticmethod
def requester_or_gifter(drift, current_user_id):
# 不建议将current_user耦合进DriftViewModel，破坏了封装性，难以扩展，所以当做参数从外部传入
return 'requester' if current_user_id == drift.requester_id else 'gifter'

def _parse(self, drift, current_user_id):
you_are = DriftViewModel.requester_or_gifter(drift, current_user_id)
# pending_status 设计到了4*2=8种状态，这个状态的判断应该在PendingStatus完成
pending_status = PendingStatus.pending_str(drift.pending, you_are)
r = {
'drift_id': drift.id,
'you_are': you_are,
'book_title': drift.book_title,
'book_author': drift.book_author,
'book_img': drift.book_img,
'date': drift.create_datetime.strftime('%Y-%m-%d'),
'operator': drift.requester_nickname if you_are != 'requester' \
else drift.gifter_nickname,
'message': drift.message,
'address': drift.address,
'status_str': pending_status,
'recipient_name': drift.recipient_name,
'mobile': drift.mobile,
'status': drift.pending
}

return r

enums.py

class PendingStatus(Enum):
Waiting = 1
Success = 2
Reject = 3
Redraw = 4

@classmethod
def pending_str(cls, status, key):
key_map = {
cls.Waiting.value: {
'requester': '等待对方邮寄',
'gifter': '等待你邮寄'
},
cls.Reject.value: {
'requester': '对方已拒绝',
'gifter': '你已拒绝'
},
cls.Redraw.value: {
'requester': '你已撤销',
'gifter': '对方已撤销'
},
cls.Success.value: {
'requester': '对方已邮寄',
'gifter': '你已邮寄，交易完成'
}
}
return key_map[status][key]

对比三种ViewModel

1.BookViewModel将单体和集合分开，单体定义实例属性，清晰明了（最推荐）
2.MyTrade 将单体作为字典融合在集合类里面，代码少，灵活性高，但是不易扩展（最不推荐）
3.DriftViewModel 将单体和集合分开，单体使用字典。坚固了前两中的优点，但是可读性下降（也不是特别推荐）

---

11.7 其他操作

1.更好的使用枚举

我们的数据库中pending存储的是数字类型，但是我们在代码中使用的是枚举类型。这肯定是匹配不上的，一种最优雅的解决方式就是为我们的Drift模型的pending属性编写getter/setter方法

# 状态
_pending = Column('pending', SmallInteger, default=1)

@property
def pending(self):
return PendingStatus(self._pending)

@pending.setter
def pending(self, status):
self._pending = status.value

这样就能在外部使用枚举类型操作我们的属性了

2.撤销操作业务逻辑

@web.route('/drift/<int:did>/redraw')
@login_required
def redraw_drift(did):
with db.auto_commit():
# 横向越权：一个用户访问另一个用户的数据并进行修改
# requester_id=current_user.id 防止横向越权
drift = Drift.query.filter_by(
id=did, requester_id=current_user.id).first_or_404()
drift.pending = PendingStatus.Redraw
current_user.beans += 1

return redirect(url_for('web.pending'))

3.拒绝操作业务逻辑

@web.route('/drift/<int:did>/redraw')
@login_required
def redraw_drift(did):
with db.auto_commit():
# requester_id=current_user.id 防止超权现象
drift = Drift.query.filter_by(
id=did, requester_id=current_user.id).first_or_404()
drift.pending = PendingStatus.Redraw
current_user.beans += 1

return redirect(url_for('web.pending'))

4.完成邮寄

@web.route('/drift/<int:did>/mailed')
def mailed_drift(did):
with db.auto_commit():
# 更改鱼漂状态位成功
drift = Drift.query.filter_by(
id=did, gifter_id=current_user.id).first_or_404()
drift.pending = PendingStatus.Success

# 赠送一个鱼豆
current_user.beans += 1

# 完成赠送
gift = Gift.query.get_or_404(drift.gift_id)
gift.launched = True

# 完成心愿
Wish.query.filter_by(
isbn=drift.isbn, uid=drift.requester_id, launched=False)\
.update({Wish.launched: True})
return redirect(url_for('web.pending'))

5.撤销赠送

@web.route('/gifts/<gid>/redraw')
def redraw_from_gifts(gid):
gift = Gift.query.filter_by(id=gid, launched=False).first_or_404()
# 思维逻辑要严谨
drift = Drift.query.filter_by(gift_id=gid, pending=PendingStatus.Waiting).first()

if drift:
flash('这个礼物正处于交易状态，请先千万鱼漂完成该交易')
else:
with db.auto_commit():
gift.delete()
current_user.beans -= current_app.config['BEANS_UPLOAD_ONE_BOOK']

return redirect(url_for('web.my_gifts'))

6.撤销心愿

@web.route('/wish/book/<isbn>/redraw')
def redraw_from_wish(isbn):
wish = Wish.query.filter_by(isbn=isbn, launched=False).first_or_404()

with db.auto_commit():
wish.delete()

return redirect(url_for('web.my_wish'))

7.赠送书籍

"""
向想要这本书的人发送一封邮件
注意，这个接口需要做一定的频率限制
这接口比较适合写成一个ajax接口
"""
wish = Wish.query.get_or_404(wid)
gift = Gift.query.filter_by(uid=current_user.id, isbn=wish.isbn).first()
if not gift:
flash('你还没有上传此书，请点击“加入到赠送清单”添加此书。添加前，请确保自己可以赠送此书')
else:
send_mail(wish.user.email, '有人想送你一本书', 'email/satisify_wish.html', wish=wish,
gift=gift)
flash('已向他/她发送了一封邮件，如果他/她愿意接受你的赠送，你将收到一个鱼漂')
return redirect(url_for('web.book_detail', isbn=wish.isbn))

email/satisify_wish.html的内容

<p><stong>亲爱的 {{ wish.user.nickname }},</stong></p>
<p>{{ gift.user.nickname }} 有一本《{{ wish.book.title }}》可以赠送给你</p>
{# 将用户导向send_drift索要数据的试图函数#}
<p>点击<a
href="{{ url_for('web.send_drift', gid=gift.id, _external=True) }}">这里</a>填写书籍邮寄地址，
等待{{ gift.user.nickname }}将书籍寄送给你
</p>
<p>如果无法点击，你也可以将下面的地址复制到浏览器中打开:</p>
<p>{{ url_for('web.send_drift', gid=gift.id, _external=True) }}</p>
<p>你的，</p>
<p>鱼书</p>
<p>
<small>注意，请不要回复此邮件哦</small>
</p>