实现导出任务
现在所有的准备工作已经完成,可以开始编写导出函数了。 这个函数的高层结构如下:
app/tasks.py :导出用户动态通用结构。
def export_posts(user_id):
try:
# read user posts from database
# send email with data to user
except:
# handle unexpected errors
为什么将整个任务包装在 try/except 块中呢? 请求处理器中的应用代码可以防止意外错误,因为 Flask 本身捕获异常,然后将它们以我设置的日志配置的方式来进行处理。 然而,这个函数将运行在由 RQ 控制的单独进程中,而非 Flask,因此如果发生任何意外错误,任务将中止,RQ 将向控制台显示错误,然后返回等待新的 job。 所以基本上,除非你正在观看 RQ worker 的输出或将其记录到文件中,否则将永远不会发现有错误。
让我们从上面带有注释的三部分中最简单的错误处理部分开始梳理:
app/tasks.py :导出用户动态错误处理。
import sys
# ...
def export_posts(user_id):
try:
# ...
except:
_set_task_progress(100)
app.logger.error('Unhandled exception', exc_info=sys.exc_info())
每当发生意外错误时,我将通过将进度设置为 100%来将任务标记为完成,然后使用 Flask 应用程序中的日志记录器对象记录错误以及堆栈跟踪信息(调用 sys.exc_info()
来获得)。 使用 Flask 应用日志记录器来记录错误的好处在于,你可以观察到你为 Flask 应用实现的任何日志记录机制。 例如,在 第七章 中,我配置了要发送到管理员电子邮件地址的错误。 只要使用 app.logger
,我也可以得到这些错误信息。
接下来,我将编写实际的导出代码,它只需发出一个数据库查询并在循环中遍历结果,并将它们累积在字典中:
app/tasks.py :从数据库读取用户动态。
import time
from app.models import User, Post
# ...
def export_posts(user_id):
try:
user = User.query.get(user_id)
_set_task_progress(0)
data = []
i = 0
total_posts = user.posts.count()
for post in user.posts.order_by(Post.timestamp.asc()):
data.append({'body': post.body,
'timestamp': post.timestamp.isoformat() + 'Z'})
time.sleep(5)
i += 1
_set_task_progress(100 * i // total_posts)
# send email with data to user
except:
# ...
每条动态都是一个包含两个条目的字典,即动态正文和动态发表的时间。 时间格式将采用 ISO 8601 标准。 我使用的 Python 的 datetime
对象不存储时区,因此在以 ISO 格式导出时间后,我添加了'Z',它表示 UTC。
由于需要跟踪进度,代码变得稍微复杂了些。 我维护了一个计数器 i
,并且在进入循环之前还需要发出一个额外的数据库查询,查询 total_posts
以获得用户动态的总数。 使用了 i
和 total_posts
,在每个循环迭代我都可以使用从 0 到 100 的数字来更新任务进度。
你可能会好奇我为什么会在每个循环迭代中加入 time.sleep(5)
调用。主要原因是我想要延长导出所需的时间,以便在用户动态不多的情况下也可以方便地查看到导出进度的增长。
下面是函数的最后部分,将会带上 data
附件发送邮件给用户:
app/tasks.py :发送带用户动态的邮件给用户。
import json
from flask import render_template
from app.email import send_email
# ...
def export_posts(user_id):
try:
# ...
send_email('[Microblog] Your blog posts',
sender=app.config['ADMINS'][0], recipients=[user.email],
text_body=render_template('email/export_posts.txt', user=user),
html_body=render_template('email/export_posts.html', user=user),
attachments=[('posts.json', 'application/json',
json.dumps({'posts': data}, indent=4))],
sync=True)
except:
# ...
其实只是对 send_email()
函数的调用。 附件被定义为一个元组,其中有三个元素被传递给 Flask-Mail 的 Message
对象的 attach()
方法。 元组中的第三个元素是附件内容,它是用 Python 的 json.dumps()
函数生成的。
这里引用了一对新模板,它们以纯文本和 HTML 格式提供电子邮件正文的内容。 这是文本模板的内容:
app/templates/email/export_posts.txt :导出用户动态文本邮件模板。
Dear {{ user.username }},
Please find attached the archive of your posts that you requested.
Sincerely,
The Microblog Team
这是 HTML 版本的邮件模板:
app/templates/email/export_posts.html :导出用户动态 HTML 邮件模板。
<p>Dear {{ user.username }},</p>
<p>Please find attached the archive of your posts that you requested.</p>
<p>Sincerely,</p>
<p>The Microblog Team</p>
如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

绑定邮箱获取回复消息
由于您还没有绑定你的真实邮箱,如果其他用户或者作者回复了您的评论,将不能在第一时间通知您!
发布评论