第21章用户通知

优质

小牛编辑

129浏览

2023-12-01

本文翻译自The Flask Mega-Tutorial Part XXI: User Notifications

这是Flask Mega-Tutorial系列的第二十一章，我将添加一个私有消息功能，它将在导航栏中显示用户通知，而且无需刷新页面就可以自动更新。

在本章中，我想继续致力于改进Microblog应用程序的用户体验。有一个广泛应用的功能是向用户显示警报或通知。社交应用通常会通过在顶部导航栏中显示带有数字的小徽章显示这些通知来让你知道有新的提及（＠）或私有消息。虽然这是最明显的用法，但通知模式还可以应用于许多其他类型的应用程序，以通知用户需要注意的事项。

为了向你展示构建用户通知所涉及的技术，我需要扩展Microblog。因此在本章的第一部分中，我将构建一个用户消息传递系统，它允许任何用户发送私有消息给另一个用户。这实际上比听起来更简单，通过它，我们可以很好地复习核心的Flask实践，并告诉你Flask到底能在简单，高效和有趣的方面做到什幺程度。一旦消息系统就位，我就会讨论一些方法来实现显示未读消息计数的通知标志。

私有消息

我要实现的私有消息功能非常简单。当你访问用户的个人主页时，会显示一个可以向该用户发送私有消息链接。该链接将带你进入一个新的页面，在新页面中，可以在Web表单中发送消息。要阅读发送给你的消息，页面顶部的导航栏将会有一个新的“消息”链接，它会将你带到与主页或发现页面相似的页面，但不会显示用户动态，它会显示其他用户发送给你的消息。

以下小节介绍了实现此功能所需的各个步骤。

私有消息的数据库支持

第一项任务是扩展数据库以支持私有消息。这是一个新的Message模型：

app/models.py：Message模型。

class Message(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    sender_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    recipient_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    body = db.Column(db.String(140))
    timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
    def __repr__(self):
        return '<Message {}>'.format(self.body)

这个模型类与Post模型相似，唯一的区别是有两个用户外键，一个用于发信人，另一个用于收信人。 User模型可以获得这两个用户的关系，以及一个新字段，用于指示用户最后一次阅读他们的私有消息的时间：

app/models.py：User模型对私有消息的支持。

class User(UserMixin, db.Model):
    # ...
    messages_sent = db.relationship('Message',
                                    foreign_keys='Message.sender_id',
                                    backref='author', lazy='dynamic')
    messages_received = db.relationship('Message',
                                        foreign_keys='Message.recipient_id',
                                        backref='recipient', lazy='dynamic')
    last_message_read_time = db.Column(db.DateTime)
    # ...
    def new_messages(self):
        last_read_time = self.last_message_read_time or datetime(1900, 1, 1)
        return Message.query.filter_by(recipient=self).filter(
            Message.timestamp > last_read_time).count()

这两个关系将返回给定用户发送和接收的消息，并且在关系的Message一侧将添加author和recipient回调引用。我之所以使用author回调而不是更适合的sender，是因为通过使用author，我可以使用我用于用户动态的相同逻辑渲染这些消息。 last_message_read_time字段将存储用户最后一次访问消息页面的时间，并将用于确定是否有比此字段更新时间戳的未读消息。 new_messages()辅助方法实际上使用这个字段来返回用户有多少条未读消息。在本章的最后，我将把这个数字作为页面顶部导航栏中的一个漂亮的徽章。

完成了数据库更改后，现在是时候生成新的迁移并使用它升级数据库了：

(venv) $ flask db migrate -m "private messages"
(venv) $ flask db upgrade

发送一条私有消息

下一步设计发送消息。我需要一个简单的Web表单来接收消息：

app/main/forms.py：私有消息表单类。

class MessageForm(FlaskForm):
    message = TextAreaField(_l('Message'), validators=[
        DataRequired(), Length(min=0, max=140)])
    submit = SubmitField(_l('Submit'))

而且我还需要在网页上呈现此表单的HTML模板：

app/templates/send_message.html：发送私有消息HTML模板。

{% extends "base.html" %}
{% import 'bootstrap/wtf.html' as wtf %}
{% block app_content %}
    <h1>{{ _('Send Message to %(recipient)s', recipient=recipient) }}</h1>
    <div class="row">
        <div class="col-md-4">
            {{ wtf.quick_form(form) }}
        </div>
    </div>
{% endblock %}

接下来，我将添加一个新的/send_message/路由来处理实际发送的私有消息：

app/main/routes.py：发送私有消息的视图函数。

from app.main.forms import MessageForm
from app.models import Message
# ...
@bp.route('/send_message/<recipient>', methods=['GET', 'POST'])
@login_required
def send_message(recipient):
    user = User.query.filter_by(username=recipient).first_or_404()
    form = MessageForm()
    if form.validate_on_submit():
        msg = Message(author=current_user, recipient=user,
                      body=form.message.data)
        db.session.add(msg)
        db.session.commit()
        flash(_('Your message has been sent.'))
        return redirect(url_for('main.user', username=recipient))
    return render_template('send_message.html', title=_('Send Message'),
                           form=form, recipient=recipient)

这个视图函数中的逻辑显而易见。发送私有消息的操作只需在数据库中添加一个新的“消息”实例即可。

将所有内容联系在一起的最后一项更改是在用户个人主页中添加上述路由的链接：

app/templates/user.html：个人主页中添加发送私有消息的链接。

                {% if user != current_user %}
                <p>
                    <a href="{{ url_for('main.send_message',
                                        recipient=user.username) }}">
                        {{ _('Send private message') }}
                    </a>
                </p>
                {% endif %}

查看私有消息

这个功能的第二大部分是查看私有信息。为此，我添加另一条路由/messages，该路由与主页和发现页面非常相似，包括分页的完全支持：

app/main/routes.py：查看消息视图函数。

@bp.route('/messages')
@login_required
def messages():
    current_user.last_message_read_time = datetime.utcnow()
    db.session.commit()
    page = request.args.get('page', 1, type=int)
    messages = current_user.messages_received.order_by(
        Message.timestamp.desc()).paginate(
            page, current_app.config['POSTS_PER_PAGE'], False)
    next_url = url_for('main.messages', page=messages.next_num) \
        if messages.has_next else None
    prev_url = url_for('main.messages', page=messages.prev_num) \
        if messages.has_prev else None
    return render_template('messages.html', messages=messages.items,
                           next_url=next_url, prev_url=prev_url)

我在这个视图函数中做的第一件事是用当前时间更新User.last_message_read_time字段。这会将发送给该用户的所有消息标记为已读。然后，我查询消息模型以获得消息列表，并按照最近的时间戳进行排序。我决定在这里复用POSTS_PER_PAGE配置项，因为用户动态和消息的页面看起来非常相似，但是如果发生了分歧，为消息添加单独的配置变量也是有意义的。分页逻辑与我用于用户动态的逻辑完全相同，因此这对你来说应该很熟悉。

上面的视图函数通过渲染一个新的/app/templates/messages.html模板文件结束，该模板如下：

app/templates/messages.html：查看消息HTML模板。

{% extends "base.html" %}
{% block app_content %}
    <h1>{{ _('Messages') }}</h1>
    {% for post in messages %}
        {% include '_post.html' %}
    {% endfor %}
    <nav aria-label="...">
        <ul class="pager">
            <li class="previous{% if not prev_url %} disabled{% endif %}">
                <a href="{{ prev_url or '#' }}">
                    <span aria-hidden="true">&larr;</span> {{ _('Newer messages') }}
                </a>
            </li>
            <li class="next{% if not next_url %} disabled{% endif %}">
                <a href="{{ next_url or '#' }}">
                    {{ _('Older messages') }} <span aria-hidden="true">&rarr;</span>
                </a>
            </li>
        </ul>
    </nav>
{% endblock %}

在这里，我采取了另一个小技巧。我注意到除了Message具有额外的recipient关系（我不需要在消息页面中显示，因为它总是当前用户），Post和Message实例具有几乎相同的结构。所以我决定复用app/templates/_post.html子模板来渲染私有消息。出于这个原因，这个模板使用了奇怪的for循环for post in messages，以便私有消息的渲染也可以套用到子模板上。

要让用户访问新的视图函数，导航页面需要生成一个新的“消息”链接：

app/templates/base.html：导航栏中的消息链接。

                    {% if current_user.is_anonymous %}
                    ...
                    {% else %}
                    <li>
                        <a href="{{ url_for('main.messages') }}">
                            {{ _('Messages') }}
                        </a>
                    </li>
                    ...
                    {% endif %}

该功能现已完成，但作为所有更改的一部分，还有一些新的文本被添加到几个位置，并且需要将这些文本合并到语言翻译中。第一步是更新所有的语言目录：

(venv) $ flask translate update

然后，app/translations中的每种语言都需要使用新翻译更新其messages.po文件。你可以在本项目的GitHub代码库中找到西班牙语翻译，或者直接下载zip文件。

静态消息通知徽章

现在私有消息功能已经实现，但是还没有通过任何渠道告诉用户有私有消息等待阅读。导航栏上的未读消息标志的最简单实现可以使用Bootstrap badge小部件渲染到基础模板中：

app/templates/base.html：导航栏的静态消息通知徽章。

                    ...
                    <li>
                        <a href="{{ url_for('main.messages') }}">
                            {{ _('Messages') }}
                            {% set new_messages = current_user.new_messages() %}
                            {% if new_messages %}
                            <span class="badge">{{ new_messages }}</span>
                            {% endif %}
                        </a>
                    </li>
                    ...

在这里，我直接从模板中调用上面添加到User模型中的new_messages()方法，并将该数字存储在new_messages模板变量中。然后，如果该变量不为零，我只需添加带有该数字的徽章到消息链接后面即可。以下是这个页面的外观：
Messages Badge

动态消息通知徽章

上一节介绍的解决方案是一种简单的常规方式来显示通知，但它有一个缺点，即徽章仅在加载新页面时刷新。如果用户花费很长时间阅读一个页面上的内容而没有点击任何链接，那幺在该时间内出现的新消息将不会显示，直到用户最终点击链接并加载新页面。

为了让这个应用程序对我的用户更有用，我希望徽章自行更新未读消息的数量，而用户不必点击链接并加载新页面。上一节的解决方案的一个问题是，当加载页面时消息计数为非零时，徽章才在页面中渲染。更方便的是始终在导航栏中包含徽章，并在消息计数为零时将其标记为隐藏。这样可以很容易地使用JavaScript显示徽章：

app/templates/base.html：使用JavaScript渲染的友好未读消息徽章。

                    <li>
                        <a href="{{ url_for('main.messages') }}">
                            {{ _('Messages') }}
                            {% set new_messages = current_user.new_messages() %}
                            <span id="message_count" class="badge"
                                  style="visibility: {% if new_messages %}visible
                                                     {% else %}hidden {% endif %};">
                                {{ new_messages }}
                            </span>
                        </a>
                    </li>

使用此版本的徽章时，我总是将其包含在内，但当new_messages非零时，visibility CSS属性设置为visible；否则设置为hidden。我还为表示徽章的元素添加了一个id属性，以便使用$('#message_count') jQuery选择器来简化这个元素的选取。

接下来，我编写一个简短的JavaScript函数，将该徽章更新为最新的数字：

app/templates/base.html：导航栏中的动态消息通知徽章

...
{% block scripts %}
    <script>
        // ...
        function set_message_count(n) {
            $('#message_count').text(n);
            $('#message_count').css('visibility', n ? 'visible' : 'hidden');
        }
    </script>
{% endblock %}

这个新的set_message_count()函数将设置徽章元素中的消息数量，并调整可见性，以便在计数为0时隐藏徽章。

向客户端发送通知

现在剩下的就是增加一种机制，通过这种机制，客户端可以定期接收有关用户拥有的未读消息数量的更新。当更新发生时，客户端将调用set_message_count()函数来使用户知道更新。

实际上有两种方法可以让服务器将这些更新告知客户端，而且你可能会猜到，这两种方法都有优点和缺点，因此选择哪种方法很大程度上取决于项目。在第一种方法中，客户端通过发送异步请求定期向服务器请求更新。来自此请求的响应是更新列表，客户端可以使用这些更新来更新页面的不同元素，例如未读消息计数标记。第二种方法需要客户端和服务器之间的特殊连接类型，以允许服务器自由地将数据推送到客户端。请注意，无论采用哪种方法，我都希望将通知视为通用实体，以便我可以扩展此框架以支持除未读消息徽章以外的其他类型的事件。

第一种解决方案最大的优点是易于实施。我需要做的只是向应用程序添加另一条路由，例如/notifications，它返回JSON格式的通知列表。然后客户端应用程序遍历通知列表并将必要的更改应用于页面。该解决方案的缺点是实际事件和通知之间会有延迟，因为客户端会定期请求通知列表。例如，如果客户端每10秒钟询问一次通知，则可能延迟10秒接收通知。

第二个解决方案需要在协议级别进行更改，因为HTTP没有服务器主动向客户端发送数据的任何规定。到目前为止，实现服务器推送消息的最常见方式是扩展服务器以支持除HTTP之外的WebSocket连接。 WebSocket是一种不同于HTTP的协议，在服务器和客户端之间建立永久连接。服务器和客户端可以随时向对方发送数据，而无需另一方请求。这种机制的优点是，无论何时发生客户感兴趣的事件，服务器都可以发送通知，而不会有任何延迟。缺点是WebSocket需要比HTTP更复杂的设置，因为服务器需要与每个客户端保持永久连接。想象一下，例如有四个worker进程的服务器通常可以服务几百个HTTP客户端，因为HTTP中的连接是短暂的并且不断被回收。而相同的服务器只能处理四个WebSocket客户端，在绝大多数情况下，这会导致资源紧张。正是由于这种限制，WebSocket应用程序通常围绕异步服务器进行设计，因为这种服务器在管理大量worker和活动连接方面效率更高。

好消息是，不管你使用什幺方法，在客户端你都会有一个回调函数，它将被更新列表调用。因此，我可以从第一个解决方案开始，该解决方案实施起来要容易得多，如果发现不足，可以迁移到WebSocket服务器，该服务器可以配置为调用相同的客户端回调。在我看来，对于这种类型的应用，第一种解决方案实际上是可以接受的。基于WebSocket的实现对于需要以接近零延迟传递更新的应用程序非常有用。

这里有一些业界的类似案例。Twitter也使用的是第一种导航栏通知的方法；Facebook使用称为长轮询的HTTP变体，它解决了直接轮询的一些限制，同时仍然使用HTTP请求；Stack Overflow和Trello这两个站点使用WebSocket来实现通知机制。你可以通过查看浏览器调试器的“Network”选项卡来查找任何网站上发生的后台活动请求。

我们继续实施轮询解决方案。首先，我要添加一个新模型来跟踪所有用户的通知，以及用户模型中的关系。

app/models.py：通知模型。

import json
from time import time
# ...
class User(UserMixin, db.Model):
    # ...
    notifications = db.relationship('Notification', backref='user',
                                    lazy='dynamic')
    # ...
class Notification(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), index=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    timestamp = db.Column(db.Float, index=True, default=time)
    payload_json = db.Column(db.Text)
    def get_data(self):
        return json.loads(str(self.payload_json))

通知将会有一个名称，一个关联的用户，一个Unix时间戳和一个有效载荷。时间戳默认从time.time()函数中获取。每种类型的通知都会有所不同，所以我将它写为JSON字符串，因为这样可以编写列表，字典或单个值（如数字或字符串）。为了方便，我添加了get_data()方法，以便调用者不必操心JSON的反序列化。

这些更改需要包含在新的数据库迁移中：

(venv) $ flask db migrate -m "notifications"
(venv) $ flask db upgrade

为了方便，我将新增的Message和Notification模型添加到shell上下文，这样我就可以直接在用flask shell命令启动的解释器中使用这两个模型了。

microblog.py: 添加Message和Notification模型到shell上下文。

# ...
from app.models import User, Post, Notification, Message
# ...
@app.shell_context_processor
def make_shell_context():
    return {'db': db, 'User': User, 'Post': Post, 'Message': Message
            'Notification': Notification}

我还将在用户模型中添加一个add_notification()辅助方法，以便更轻松地处理这些对象：

app/models.py：Notification模型。

class User(UserMixin, db.Model):
    # ...
    def add_notification(self, name, data):
        self.notifications.filter_by(name=name).delete()
        n = Notification(name=name, payload_json=json.dumps(data), user=self)
        db.session.add(n)
        return n

此方法不仅为用户添加通知给数据库，还确保如果具有相同名称的通知已存在，则会首先删除该通知。我将要使用的通知将被称为unread_message_count。如果数据库已经有一个带有这个名称的通知，例如值为3，则当用户收到新消息并且消息计数变为4时，我就会替换旧的通知。

在任何未读消息数改变的地方，我需要调用add_notification()，以便我更新用户的通知，这样的地方有两处。首先，在send_message()视图函数中，当用户收到一个新的私有消息时：

app/main/routes.py：更新用户通知。

@bp.route('/send_message/<recipient>', methods=['GET', 'POST'])
@login_required
def send_message(recipient):
    # ...
    if form.validate_on_submit():
        # ...
        user.add_notification('unread_message_count', user.new_messages())
        db.session.commit()
        # ...
    # ...

第二个地方是用户转到消息页面时，未读计数需要归零：

app/main/routes.py：查看消息视图函数。

@bp.route('/messages')
@login_required
def messages():
    current_user.last_message_read_time = datetime.utcnow()
    current_user.add_notification('unread_message_count', 0)
    db.session.commit()
    # ...

既然用户的所有通知都保存在数据库中，那幺我可以添加一条新路由，客户端可以使用该路由为登录用户检索通知：

app/main/routes.py：通知视图函数。

from app.models import Notification
# ...
@bp.route('/notifications')
@login_required
def notifications():
    since = request.args.get('since', 0.0, type=float)
    notifications = current_user.notifications.filter(
        Notification.timestamp > since).order_by(Notification.timestamp.asc())
    return jsonify([{
        'name': n.name,
        'data': n.get_data(),
        'timestamp': n.timestamp
    } for n in notifications])

这是一个相当简单的函数，它返回一个包含用户通知列表的JSON负载。每个通知都以包含三个元素的字典的形式给出，即通知名称，与通知有关的附加数据（如消息数量）和时间戳。通知按照从创建时间顺序进行排序。

我不希望客户重复发送通知，所以我给他们提供了一个选项，只请求给定时间戳之后产生的通知。 since选项可以作为浮点数包含在请求URL的查询字符串中，其中包含开始时间的unix时间戳。如果包含此参数，则只有在此时间之后发生的通知才会被返回。

完成此功能的最后一部分是在客户端实现实际轮询。最好的做法是在基础模板中实现，以便所有页面自动继承该行为：

app/templates/base.html：轮询通知。

...
{% block scripts %}
    <script>
        // ...
        {% if current_user.is_authenticated %}
        $(function() {
            var since = 0;
            setInterval(function() {
                $.ajax('{{ url_for('main.notifications') }}?since=' + since).done(
                    function(notifications) {
                        for (var i = 0; i < notifications.length; i++) {
                            if (notifications[i].name == 'unread_message_count')
                                set_message_count(notifications[i].data);
                            since = notifications[i].timestamp;
                        }
                    }
                );
            }, 10000);
        });
        {% endif %}
    </script>

该函数包含在一个模板条件中，因为我只想在用户登录时轮询新消息。对于没有登录的用户，这个函数将不会被渲染。

你已经在第二十章中看到了jQuery的$(function() { ...})模式。这是注册一个函数在页面加载后执行的方式。对于这个功能，我需要在页面加载时做的是设置一个定时器来获取用户的通知。你还看到了setTimeout() JavaScript函数，它在等待特定时间之后运行作为参数给出的函数。 setInterval()函数使用与setTimeout()相同的参数，但不是一次性触发定时器，而是定期调用回调函数。本处，我的间隔设置为10秒（以毫秒为单位），所以我将以每分钟大约六次的频率查看通知是否有更新。

利用定期计时器和Ajax，该函数轮询新通知路由，并在其完成回调中迭代通知列表。当收到名为unread_message_count的通知时，通过调用上面定义的函数和通知中给出的计数来调整消息计数徽章。

我处理since参数的方式可能会令人困惑。我首先将这个参数初始化为0。参数总是包含在请求URL中，但是我不能像以前那样使用Flask的url_for()来生成查询字符串，因为一次请求中url_for()只在服务器上运行一次，而我需要since参数动态更新多次。第一次，这个请求将被发送到/notifications?since=0，但是一旦我收到通知，我就会将since更新为它的时间戳。这可以确保我不会收到重复的内容，因为我总是要求收到自我上次看到的通知以来发生的新通知。同样重要的是要注意，我在interval函数外声明since变量，因为我不希望它是局部变量，我想要在所有调用中使用相同的变量。

最简单的测试方法是使用两种不同的浏览器A和B。在两个浏览器上使用不同的用户登录Microblog。然后从A浏览器向B浏览器上的用户发送一个或多个消息。 B浏览器的导航栏应更新为显示你在10秒钟内发送的消息数量。而当你点击消息链接时，未读消息数重置为零。

第21章 用户通知