跳转到主要内容

Grok消息机制

项目描述

grokcore.message

此包为grok设置提供z3c.flashmessage的集成。这意味着需要处理:

  • 在组件架构中注册一个全局消息接收器。

  • 默认注册一个名为“session”的全局基于会话的消息源。

  • 可选地(如果包含“ram.zcml”),注册一个名为“ram”的全局RAM存储消息源。

  • 提供组件以利用全局消息接收器和源。

有关我们在这里讨论的消息类型的详细信息,请参阅z3c.flashmessage文档。

设置grokcore.message

当被grok处理时,grokcore.message注册:

  • 一个名为“session”的全局会话消息源。

  • 一个全局消息接收器。

此包的grok处理发生在执行本地configure.zcml时。在基于Grok的标准包中,这通常自动发生。

当然,也可以手动grok这个包

>>> import grokcore.component as grok
>>> grok.testing.grok('grokcore.message')

这设置了一个全局消息接收器

>>> from z3c.flashmessage.interfaces import IMessageReceiver
>>> from zope.component import getUtility
>>> getUtility(IMessageReceiver)
<z3c.flashmessage.receiver.GlobalMessageReceiver object at 0x...>

它还设置了一个名为“session”的基于会话的消息源

>>> from z3c.flashmessage.interfaces import IMessageSource
>>> getUtility(IMessageSource, name=u'session')
<z3c.flashmessage.sources.SessionMessageSource object at 0x...>

我们还提供了一个可以由包含“ram.zcml”来启用且默认未注册的RAM存储消息源

>>> getUtility(IMessageSource, name=u'ram')
Traceback (most recent call last):
...
zope.interface.interfaces.ComponentLookupError: (<InterfaceClass z3c.flashmessage.interfaces.IMessageSource>, 'ram')

您可以通过在ZCML设置中包含来自“grokcore.message”的“ram.zcml”来启用此源,如下所示

<configure xmlns="http://namespaces.zope.org/zope">
  <include package="grokcore.message" file="ram.zcml" />
</configure>

或者,当然,通过手动注册RAMMessageSource

>>> from zope.component import provideUtility
>>> from z3c.flashmessage.sources import RAMMessageSource
>>> ram_source = RAMMessageSource()
>>> provideUtility(ram_source, name=u'ram')

现在我们可以获取RAM源

>>> getUtility(IMessageSource, name=u'ram')
<z3c.flashmessage.sources.RAMMessageSource object at 0x...>

组件(API)

grokcore.message除了从z3c.flashmessage提供的常规组件外,还提供了一些额外的组件和函数。

UniqueMessageSource

UniqueMessageSource是一个恰好包含零个或一个消息的消息源。请注意,消息不会在UniqueMessageSource实例中持久存储,并且在重启您的Zope实例后将会丢失。

它是一个基类,这意味着您必须从它派生出来,以便在软件被grok处理后将其实例注册为全局实用程序(请参见下面的示例)。

方法

UniqueMessageSource.send(message[, type=u’message’])

发送类型为type的消息message

UniqueMessageSource.list(type=None)

返回一个生成器对象,列出存储的消息(如果有)。

UniqueMessageSource.delete(message)

如果message是存储的消息,则从源中删除存储的消息。

便捷函数

grokcore.message提供了一些便捷函数来向源提供数据或从源获取数据。

grokcore.message.send(message[, type=’message’[, name=’session’]])

message发送到名为name的消息源。

如果消息可以成功发送,则返回True。否则返回False

>>> import grokcore.message
>>> grokcore.message.send('Meet at dawn!')
True
>>> grokcore.message.send('Meat a fawn!', name='doesnotexist')
False

grokcore.message.get_from_source([name=’’])

获取在名为name的注册消息源中存储的消息列表或None

此操作不会从查询的源中删除消息。

>>> import grokcore.message
>>> grokcore.message.get_from_source('session')
<generator object ...>
>>> grokcore.message.get_from_source('not-existing') is None
True

grokcore.message.receive([name=’’])

接收名为name的注册接收器收集的消息。

>>> import grokcore.message
>>> msgs = list(grokcore.message.receive())
>>> msgs
[<z3c.flashmessage.message.Message object at 0x...>]
>>> msgs[0].message
'Meet at dawn!'

请注意,此操作可能会从它们被发送到的源中删除消息,因为通过“接收”消息,您表示消息已被处理。

例如,会话源现在为空

>>> list(grokcore.message.get_from_source('session'))
[]

再次接收将不会给出任何结果

>>> list(grokcore.message.receive())
[]

示例

创建一个 UniqueMessageSource

>>> from grokcore.message import UniqueMessageSource
>>> class MyUniqueMessageSource(UniqueMessageSource):
...   grok.name('uniq_source')

在解析后,源将自动注册

>>> grok.testing.grok_component(
...     'MyUniqueMessageSource', MyUniqueMessageSource,
...     dotted_name='grokcore.message.tests')
True
>>> source = getUtility(IMessageSource, name='uniq_source')
>>> source
<...MyUniqueMessageSource object at 0x...>

它提供了由 IMessageSource 接口要求的方法

>>> from z3c.flashmessage.interfaces import IMessageSource
>>> from zope.interface import verify
>>> verify.verifyClass(IMessageSource, MyUniqueMessageSource)
True

我们可以列出存储在源中的消息

>>> source.list()
<generator object ...>
>>> list(source.list())
[]
>>> source.send(message='Hello!', type='message')
>>> list(source.list())
[<z3c.flashmessage.message.PersistentMessage object at 0x...>]
>>> print(list(source.list())[0].message)
Hello!

当我们发送另一条消息时,旧的消息将被默默地丢弃

>>> source.send(message='Hello again!', type='message')
>>> len(list(source.list()))
1
>>> print(list(source.list())[0].message)
Hello again!

我们可以删除消息

>>> msg = list(source.list())[0]
>>> source.delete(msg)
>>> len(list(source.list()))
0

方便函数的示例可以在上方找到。

CHANGES

4.0 (2023-08-28)

  • 停止支持 Python 2.7, 3.4, 3.5, 3.6。

  • 添加对 Python 3.7, 3.8, 3.9, 3.10, 3.11 的支持。

3.0.1 (2018-01-17)

  • 在整个文档中用 @grok.implementer() 指令替换 grok.implements() 的使用。

3.0.0 (2018-01-15)

  • Python 3 兼容性。

0.4.3 (2016-02-15)

  • 更新测试。

0.4.2 (2010-10-25)

  • 通过显式注册 IClientIdManager 和 ISessionDataContainer 工具修复了测试。为此重新引入了 ftesting.zcml。

0.4.1 (2010-10-25)

  • 删除不再必要的 ftesting.zcml。

0.4 (2010-10-25)

  • 确保 zope.session 已配置,因为该包声称提供基于会话的闪存消息机制。

  • 使包符合 zope.org 存储库策略。

0.3 (2010-03-05)

  • UniqueMessageSource 现在完全实现了 IMessageSource 接口,即使用 UniqueMessageSource.send() 时,type 参数现在是可选的。

0.2 (2010-03-03)

  • 实用函数 send 现在接收一个 name 参数,允许选择目标消息源。

0.1 (2010-03-03)

  • grokui.admingrokmegrok.layout 的旧版本中提取出来。

项目详情


下载文件

为您的平台下载文件。如果您不确定选择哪个,请了解有关 安装包 的更多信息。

源分布

grokcore.message-4.0.tar.gz (10.1 kB 查看散列)

上传时间

构建分布

grokcore.message-4.0-py3-none-any.whl (11.8 kB 查看散列)

上传时间 Python 3

由以下支持

AWS AWS 云计算和安全赞助商 Datadog Datadog 监控 Fastly Fastly CDN Google Google 下载分析 Microsoft Microsoft PSF 赞助商 Pingdom Pingdom 监控 Sentry Sentry 错误记录 StatusPage StatusPage 状态页面