“promisify”(异步/等待)使用 Python 中的 pymsteams 向 Microsoft Teams 发送消息

查波

我正在使用该logging模块在我的 Python 脚本中向 MS Teams 发送大量消息不幸的是,这很慢,所以我想向消息添加异步/等待功能。

这是我的记录器模块(有些简化):

from logging import StreamHandler
import pymsteams

class TeamsHandler(StreamHandler):

    def __init__(self, channel_url):
        StreamHandler.__init__(self)
        self.channel_url = channel_url
        self.client = pymsteams.connectorcard(self.channel_url)

    def emit(self, record):
        msg = self.format(record)
        self.client.text(msg)
        try:
            self.client.send()
        except:
            print(f"{msg} could not be sent to Teams")

然后您将在常规脚本中使用它:

import logging
from TeamsHandler import TeamsHandler #the module above

my_logger = logging.getLogger('TestLogging')
my_logger.setLevel(logging.DEBUG)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
my_logger.addHandler(console_handler)

CHANNEL_ID = "https://outlook.office.com/webhook/your-big-number"
teamshandler = TeamsHandler(CHANNEL_ID)
teamshandler.setFormatter(logging.Formatter('%(levelname)s %(message)s'))
teamshandler.setLevel(logging.DEBUG)
my_logger.addHandler(teamshandler)
for i in range(1,100):
    my_logger.error(f"this is an error [{i}]")
    my_logger.info(f"this is an info [{i}]")
do_something_else()

我怎样才能让它do_something_else立即(几乎)执行,而不必等待 200 条消息才能进入 Teams?

我尝试添加一些asyncawait在中的关键字TeamsHandler模块,但不尝试是成功的,所以我没有把他们的问题。如果不是完整的解决方案,很高兴得到一些指导。

理想情况下,随着脚本的进行,消息的顺序应该保持不变。

用户4815162342

如果pymsteams不支持 async/await,那么添加async到您的函数不会真正帮助您,因为您最终仍会从pymsteams. 即使它确实支持 async/await,它仍然无法工作,因为您是从 Python 日志记录 API 内部调用它们的,而 Python 日志记录 API 本身不是异步的。async/await 不能神奇地将同步代码转换为 async,程序必须全面使用 async/await。

但是,如果您需要异步执行只是在后台运行某些东西,则可以改用线程。例如,创建一个专门的线程进行日志记录,例如:

class TeamsHandler(StreamHandler):
    def __init__(self, channel_url):
        super().__init__()
        self.channel_url = channel_url
        self.client = pymsteams.connectorcard(self.channel_url)
        self.queue = queue.Queue()
        self.thread = threading.Thread(target=self._worker)
        self.thread.start()
        # shutdown the worker at process exit
        atexit.register(self.queue.put, None)

    def _worker(self):
        while True:
            record = self.queue.get()
            if record is None:
                break
            msg = self.format(record)
            self.client.text(msg)
            try:
                self.client.send()
            except:
                print(f"{msg} could not be sent to Teams")

    def emit(self, record):
        # enqueue the record to log and return control to the caller
        self.queue.put(record)

当然,这有一个缺点,如果日志后端出现问题,您的程序可能会比您在日志中看到的内容提前——但是当您删除日志记录和执行之间的同步时,情况总是如此。

本文收集自互联网,转载请注明来源。

如有侵权,请联系 [email protected] 删除。

编辑于
0

我来说两句

0 条评论
登录 后参与评论

相关文章

如何使用Python向Viber机器人发送消息?

在Python Flask中从Microsoft Teams bot验证HMAC

在Telethon Python电报中向联系人发送消息

Chatbot在Microsoft Teams中未发送回消息

如何在ASP.NET Core中验证从Microsoft Teams发送的消息有效载荷?

使用WebHook将文件发送到Microsoft Teams

通过传入的Webhook向Microsoft Teams发送对较旧消息的答复

通过向通道发送文件信息,Microsoft Teams Bot文件卡信息已损坏

如何使用pymsteams发送df变量的结果(python ms团队)

如何在Azure中向Microsoft Teams Bot添加授权

向Microsoft Teams消息添加动态链接

向Microsoft Teams发送对话主动消息时,clientSecret不能为null

从SSIS发送Microsoft Teams聊天消息的方法

使用Graph API在Microsoft Teams的机器人通道中以机器人的身份向用户发送消息

可用于向Microsoft Teams中的Planner应用添加任务的任何API

使用Python的Microsoft Teams机器人

是否可以向Microsoft Teams消息添加反应?

使用Java中的JButton向客户端发送消息

使用Python中的PyObjC和ScriptingBridge在消息中发送消息

如何使用 python 中的 amqp 向 Azure 事件中心发送消息

在 Microsoft Teams 中通过移动设备上的消息启动操作?

使用适用于 Python 的 Bot Framework SDK v4 初始化并向 Microsoft Teams 频道发送消息

使用 Microsoft Graph API 或 BOT API 发送 MS Teams 消息

在 Microsoft Teams 中从空频道获取消息导致 403

如何使用 JavaScript 从 HTML 向 Python (Flask) 发送消息?

使用 Bot 框架在 Microsoft Teams 上发送主动消息

如何使用 Microsoft Graph 从 Microsoft Teams 机器人发送消息?

使用 Python 向 Slack 发送带有附件的消息

在 Openshift 中向 pod 发送 REST 消息