消息通知
通知
Masonite 有一个简单而强大的通知功能,用于从您的应用程序发送通知。以下是您可以对通知执行的操作的简要概述:
- 发送
E-mail
、Slack
和SMS
通知 - 将通知存储在数据库中,以便它们可以显示在您的 Web 界面中。
- 队列通知
- 广播通知
创建通知
要使用 Masonite 创建和发送通知,您必须首先构建一个Notification
类。此类将充当您的通知设置,例如传递渠道和这些不同渠道(邮件、slack、短信...)的通知内容。
构建通知的第一步是运行命令:
$ python craft notification Welcome
这将创建您的通知,它看起来像这样:
class Welcome(Notification, Mailable):
def to_mail(self, notifiable):
return (
self.to(notifiable.email)
.subject("Welcome to our site!")
.from_("admin@example.com")
.text(f"Hello {notifiable.name}")
)
def via(self, notifiable):
return ["mail"]
每个通知类都有一个via
方法,用于指定通知将在哪些通道上传递。通知可以通过mail
、database
、broadcast
、slack
和vonage
频道发送。稍后将对此进行更多详细说明。发送通知时,它将自动发送到每个频道。
提示:如果您想使用其他交付渠道,请随时检查是否为其开发了社区驱动程序或创建自己的驱动程序并与社区共享!
via
方法应返回您希望发送通知的频道列表。此方法接收notifiable
实例。
class WelcomeNotification(Notification, Mailable):
...
def via(self, notifiable):
"""通过电子邮件发送此通知,Slack 并将其保存在数据库中。"""
return ["mail", "slack", "database"]
发送通知
基础
你可以使用 Notification
类轻松地在控制器内部发送通知:
from masonite.notification import NotificationManager
from app.notifications.Welcome import Welcome
class WelcomeController(Controller):
def welcome(self, notification: NotificationManager):
notification.route('mail', 'sam@masonite.com').send(Welcome())
如果需要将通知传递到多个通道,可以连接不同的路由:
notification.route('mail', 'sam@masonite.com').route('slack', '#general').send(Welcome())
注意:
database
通道不能与这些通知一起使用,因为它没有附加通知实体。
通知
如果你想发送通知,例如对于你的应用程序中的用户,你可以将它们定义为通知文件。然后你仍然可以使用 Notification
类来发送通知:
from masonite.notification import NotificationManager
from app.notifications.Welcome import Welcome
class WelcomeController(Controller):
def welcome(self, notification: NotificationManager):
user = self.request.user()
notification.send(user, Welcome())
# 发送给所有用户
users = User.all()
notification.send(users, Welcome())
或者使用方便的 notify()
方法:
user = self.request.user()
user.notify(Welcome())
使用通知
ORM 模型可以定义为 Notifiable
以允许向它们发送通知。最常见的用例是将 User
模型设置为 Notifiable
因为我们经常需要向用户发送通知。
要将模型设置为可通知的 (Notifiable),添加 Notifiable
混合(mixin):
from masonite.notification import Notifiable
class User(Model, Notifiable):
# ...
你现在可以使用 user.notify()
方法向它发送通知。
路由
然后,你可以定义应如何为不同通道进行路由,这定义在 route_notification_for_{driver}
方法内。
例如,使用 mail
驱动,你可以定义:
from masonite.notification import Notifiable
class User(Model, Notifiable):
...
def route_notification_for_mail(self):
return self.email
这实际上是邮件驱动程序的默认行为,因此你不需要重复编写,但如果你的用户模型没有 email
字段或者你想添加一些字段,则可以根据需要自定义获取收件人电子邮件的逻辑。
队列通知
如果你想将通知排队,那么只需要继承 Queueable
类,它会自动将你的通知发送到队列中以供稍后处理。这是加速应用程序的好方法:
from masonite.notification import Notification
from masonite.queues import Queueable
class Welcome(Notification, Queueable):
# ...
通道
邮件
你应该在通知类上定义一个 to_mail
方法来指定如何构建电子邮件的通知内容。
class Welcome(Notification, Mailable):
def to_mail(self, notifiable):
return (
self.to(notifiable.email)
.subject("Welcome to our site!")
.from_("admin@example.com")
.text(f"Hello {notifiable.name}")
)
def via(self, notifiable):
return ["mail"]
通知将使用 config/mail.py
中定义的默认邮件驱动程序发送。有关构建邮件通知的选项的更多信息,
请查看 可邮寄选项。
如果你想覆盖给定通知的邮件驱动程序,可以执行以下操作:
class Welcome(Notification, Mailable):
def to_mail(self, notifiable):
return (
self.to(notifiable.email)
.subject("Welcome to our site!")
.from_("admin@example.com")
.text(f"Hello {notifiable.name}")
.driver("mailgun")
)
def via(self, notifiable):
return ["mail"]
Slack
你应该在通知类上定义一个 to_slack
方法来指定如何构建 slack 通知内容。
from masonite.notification.components import SlackComponent
class Welcome(Notification):
def to_slack(self, notifiable):
return SlackComponent().text('Masonite Notification: Read The Docs!, https://docs.masoniteproject.com/') \
.channel('#bot') \
.as_user('Masonite Bot') \
def via(self, notifiable):
return ["slack"]
可选项
SlackComponent
采用不同的选项来配置你的通知:
方法 | 描述 | 样例 |
---|---|---|
.text() | 你想在消息中显示的文字 | .text('欢迎来到 Masonite!') |
.to() | 你要广播到的频道。如果提供的值以 # 符号开头,则 Notifications 将使用你的令牌向 Slack 频道列表 API 发出 POST 请求并获取频道 ID。如果你不想要这种行为,则可以直接指定频道 ID | .to('#general') .to('CHSUU862') |
.send_from() | 你要显示为消息发件人的用户名。你还可以指定将显示为发件人的 url 或 icon |
.send_from('Masonite Bot', icon=":ghost:") |
.as_current_user() | 这会将一个布尔值设置为 True 消息以确定是否应显示给当前经过身份验证的用户 | .as_current_user() |
.link_names() | 这可以在消息中查找和链接频道名称和用户名 | .link_names() |
.can_reply() | 这授权回复消息 | .can_reply() |
.without_markdown() | 这不会解析消息中的任何 markdown。这是一个布尔值,不需要参数 | .without_markdown() |
.unfurl_links() | 这可以显示消息附件和链接预览。通常 slack 在发布链接时会显示网站的图标, 这将为当前消息启用该功能 | .unfurl_links() |
.as_snippet() | 用于将当前消息作为片段而不是普通消息发布。此选项有 3 个关键字参数。file_type 、name 和 title , 使用不同的 API ,因此可能无法使用以前的一些方法 |
.as_snippet(file_type='python', name='snippet', title='Awesome Snippet') |
.token() | 覆盖全局配置的令牌 (token) | .token('xoxp-359926262626-35...') |
在 Masonite 中,通知可以通过两种方式发送到 Slack 工作区:
传入 webhooks
你需要为你的 Slack 工作区 配置“传入 Webhook” 集成。此集成将为你提供在路由 Slack 通知时可以使用的 URL。此 URL 将针对特定的 Slack 通道。
Web API
你将需要 生成令牌(token) 与 Slack 工作区进行交互。
提示:这个令牌应该至少有
channels:read
,chat:write:bot
,chat:write:user
和files:write :user
权限。如果你的令牌没有这些权限,则此功能的某些部分将不起作用。
然后,你可以在 config/notifications.py
文件中将这个令牌全局定义为 SLACK_TOKEN
环境变量。或者你可以为每个通知配置不同的令牌(最终具有不同的范围)。
高级格式化
Slack 通知可以使用 Slack Blocks Kit 来构建更复杂的通知。在使用它之前,只需要安装 slackblocks
python API 来处理 Block Kit 格式。
$ pip install slackblocks
然后,你可以导入 Slack 文档中可用的大部分块并开始构建您的通知。你需要使用 block()
选项,可以根据需要链接任意数量的 block。
from masonite.notification.components import SlackComponent
from slackblocks import HeaderBlock, ImageBlock, DividerBlock
class Welcome(Notification):
def to_slack(self, notifiable):
return SlackComponent() \
.text('Notification text') \
.channel('#bot') \
.block(HeaderBlock("Header title")) \
.block(DividerBlock()) \
.block(ImageBlock("https://path/to/image", "Alt image text", "Image title"))
你可以在 slackblocks
文档 中找到所有块名称和选项,在 Slack 块列表中获取更多信息。
注意:
slackblocks
中可能还没有一些块或元素,但大部分应该都存在。
路由到通知
你应该在你的通知上定义相关的 route_notification_for_slack
方法以返回以下之一
- webhook URL 或 webhook URL 列表(如果使用 Incoming Webhooks)
- 频道名称/ID 或频道名称/ID 列表(如果使用 Slack Web API)
class User(Model, Notifiable):
def route_notification_for_slack(self):
"""Examples for Incoming Webhooks."""
# 一个 webhook
return "https://hooks.slack.com/services/..."
# 多个 webhooks
return ["https://hooks.slack.com/services/...", "https://hooks.slack.com/services/..."]
class User(Model, Notifiable):
def route_notification_for_slack(self):
"""Examples for Slack Web API."""
# 一个通道名
return "#general"
# 多个通道名
return ["#users", "#general"]
# 一个通道 ID
return "C1234567890"
路由到匿名(无通知主体)
要在没有通知实体的情况下发送 Slack 通知,必须使用 route
方法
notification.route("slack", "#general").notify(Welcome())
第二个参数可以是 频道名称
、频道ID
或 webhook URL
。
注意:指定频道名称时,必须在名称中保留
#
,如示例中所示。基于此名称,将进行反向查找以找到相应的 Slack 通道 ID。如果你想避免这个额外的调用,你可以在你的 Slack 工作区中获取频道 ID(右键单击 Slack 频道 > 复制名称 > ID 在 url 的末尾)
SMS
在 Masonite 中发送 SMS 通知由 Vonage(以前称为 Nexmo)提供支持。在通过 Vonage 发送通知之前,你需要安装 vonage
Python 客户端。
$ pip install vonage
然后你应该在 notifications.py
配置文件中配置 VONAGE_KEY
和 VONAGE_SECRET
凭证:
# config/notifications.py
VONAGE = {
"key": env("VONAGE_KEY"),
"secret": env("VONAGE_SECRET"),
"sms_from": "1234567890"
}
你还可以(全局)定义 sms_from
,它是发送短信的电话号码或姓名。你可以在 Vonage 仪表板中为你的应用程序生成电话号码。
你应该在通知类上定义一个 to_vonage
方法来指定如何构建短信通知内容。
from masonite.notification.components import Sms
class Welcome(Notification):
def to_vonage(self, notifiable):
return Sms().text("Welcome!")
def via(self, notifiable):
return ["vonage"]
选项
如果短信通知中包含 unicode 字符,构造通知时应调用 unicode 方法
Sms().text("Welcome unicode message!").set_unicode()
全局 sms_from
编号可以在通知类中被覆盖:
Sms().text("Welcome!").sms_from("+123 456 789")
路由到通知
你应该在你的通知上定义相关的 route_notification_for_vonage
方法以返回一个电话号码或一个电话号码列表来发送通知。
class User(Model, Notifiable):
def route_notification_for_vonage(self):
return self.phone
# or return [self.mobile_phone, self.land_phone]
路由到匿名实体
要在没有通知实体的情况下发送 SMS 通知,你必须使用 route
方法
notification.route("vonage", "+33612345678").notify(Welcome())
保存通知
当发送到 Notifiable
实体时,通知可以存储在你的应用程序数据库中。通知存储在 notificiation
表中。此表将包含通知类型等信息以及描述通知的 JSON 数据结构。
要将通知存储在数据库中,你应该在通知类上定义一个 to_database
方法来指定如何构建将被持久化的通知内容。
class Welcome(Notification):
def to_database(self, notifiable):
return {"data": "Welcome {0}!".format(notifiable.name)}
def via(self):
return ["mail", "database"]
此方法应返回 str
、dict
或 JSON
数据(因为它将保存到通知表中的 TEXT
列中)。你还需要将 database
通道添加到 via
方法以启用数据库通知存储。
初始化设定
在你可以将通知存储在数据库中之前,你必须创建数据库“通知”表。
$ python craft notification:table
然后你可以迁移你的数据库
$ python craft migrate
描述通知的 ORM 模型是 DatabaseNotification
并具有以下字段:
id
是模型的主键(用 UUID4 定义)type
将通知类型存储为字符串(例如WelcomeNotification
)read_at
是指示何时读取通知的时间戳data
是to_database()
的序列化表示notifiable
是返回通知所属的Notifiable
实体的关系(例如User
)created_at
,updated_at
时间戳
查询通知
通知实体具有 notifications
关系,该关系将返回该实体的通知:
user = User.find(1)
user.notifications.all() # == Collection of DatabaseNotification belonging to users
你可以直接获取未读/已读通知:
user = User.find(1)
user.unread_notifications.all() # == 用户未读 DatabaseNotification 的集合
user.read_notifications.all() # == 用户读取 DatabaseNotification 的集合
管理通知
你可以使用以下 mark_as_read
和 mark_as_unread
方法将通知标记为已读或未读
user = User.find(1)
for notification in user.unread_notifications.all():
notification.mark_as_read()
提示:最后,请记住,数据库通知可以用作任何
Masonite ORM
模型,这意味着你可以直接在模型上进行更复杂的查询以获取通知。
from masonite.notification import DatabaseNotification
DatabaseNotification.all()
DatabaseNotification.where("type", "WelcomeNotification")
广播通知
如果你想广播通知,那么你需要:
- 继承
CanBroadcast
类并指定broadcast_on
方法 - 在通知类上定义一个
to_broadcast
方法来指定如何构建将被广播的通知内容
class Welcome(Notification, CanBroadcast):
def to_broadcast(self, notifiable):
return f"Welcome {notifiable.name} !"
def broadcast_on(self):
return "channel1"
def via(self, notifiable):
return ["broadcast"]
广播到通知
默认情况下,通知将广播到在 broadcast_on
方法中定义的频道,但你可以通过在你的通知上实现 route_notification_for_broadcast
方法来覆盖每个通知:
class User(Model, Notifiable):
def route_notification_for_broadcast(self):
return ["general", f"user_{self.id}"]
匿名广播
notification.route("broadcast", "channel1").notify(Welcome())
添加新驱动
Masonite 附带了一些通知通道,但你可能希望编写自己的驱动程序以通过其他通道传递通知。 Masonite 使这个过程变得简单。
创建驱动程序
为了创建通知驱动程序,需要实现两种方法:send
和 queue
。
from masonite.notification.drivers import BaseDriver
class VoiceDriver(BaseDriver):
def send(self, notifiable, notification):
"""Specify here how to send the notification with this driver."""
data = self.get_data("voice", notifiable, notification)
# do something
get_data()
方法将可用,并将返回通知的 to_voice()
方法中定义的数据。
注册驱动
作为任何驱动程序,它应该通过自定义提供程序进行注册,例如:
from masonite.providers import Provider
class VoiceNotificationProvider(Provider):
def register(self):
self.application.make("notification").add_driver("voice", VoiceDriver(self.application))
然后,你可以将此代码脚手架到一个新的 Masonite 包中,以便社区可以使用它 😉 !
高级用法
试运行
你可以启用 dry
通知以避免发送通知。它在某些情况下很有用(后台任务、生产命令、开发、测试......)
user.notify(WelcomeNotification(), dry=True)
# or
notification.send(WelcomeNotification(), dry=True)
忽略错误
当启用 fail_silently
参数时,如果发生错误,通知发送不会引发异常。
user.notify(WelcomeNotification(), fail_silently=True)
# or
notification.send(WelcomeNotification(), fail_silently=True)
覆盖通道
通知的 via()
方法中定义的通道可以在发送时被覆盖:
class Welcome(Notification):
def to_mail(self, notifiable):
#...
def to_slack(self, notifiable):
#...
def to_database(self, notifiable):
#...
def via(self):
"""Default behaviour is to send only by email."""
return ["mail"]
使用 channels
参数,我们可以发送到其他通道(如果在通知类中正确定义):
user.notify(Welcome(), channels=["slack", "database"])
# or
notification.send(Welcome(), channels=["slack", "database"])
本译文仅用于学习和交流目的,转载请务必注明文章译者、出处、和本文链接
我们的翻译工作遵照 CC 协议,如果我们的工作有侵犯到您的权益,请及时联系我们。