有没兴趣开发个async版本的么

[XieXiaonan]

非常感谢分享
但在使用过程当中发现每次发完消息之后代码都卡住好久了,然后才发现原来是代码没用异步,
因此建议下大佬可以试试async

[whg517]

You can do this:

import asyncio
from asyncio import AbstractEventLoop
from typing import Optional

from dingtalkchatbot.chatbot import DingtalkChatbot

from crawlers_customsdata.config import settings


class DingBot:
    """Ding bot"""

    def __init__(
            self,
            token: str,
            loop: Optional[AbstractEventLoop] = None
    ):
        """
        :param token:
        :param loop:
        """
        self._loop = loop or asyncio.get_event_loop()
        self.bot = DingtalkChatbot(f'https://oapi.dingtalk.com/robot/send?access_token={token}')

    @property
    def loop(self) -> AbstractEventLoop:
        return self._loop

    async def send(self, title: str, text: str) -> None:
        """
        Send something to bot
        :param title: str
        :param text: str
        :return: None
        """
        await self.loop.run_in_executor(None, self.bot.send_markdown, title, text)


async def main() -> None:
    """unit test"""
    bot = DingBot(settings.DINGTALK_TOKEN)
    await bot.send('Hello world', '### Hello world\n'
                                  '> I sand massage to ding talk. Wonderful!!!'
                   )


if __name__ == '__main__':
    asyncio.run(main())

You can wrap the sync method around concurrent.futures.ThreadPoolExecutor . And you can read Executing code in thread or process pools to get more info.

[XieXiaonan]

非常感谢,我去学习学习