❤️ cardaio

一个用于可调整延迟的Python库。

主要用例是在你想定期从第三方系统获取某些内容，但你不知道系统将如何经常更新时。例如，你在监控一个新闻网站上的特定主题，并希望在新闻可用时立即获取新闻，同时在不更新时不要发送太多请求。

😎 功能

• 🐍 纯Python。
• 🐎 以asyncio为第一选择，同时提供同步API。
• 🛡 完全类型注解和类型安全。
• 🪶 无依赖。
• 🔧 高度可配置。
• ☑️ 100%测试覆盖率。

📦 安装

python3 -m pip install cardaio

🛠️ 使用

from cardaio import Heartbeat

async def run() -> None:
    hb = Heartbeat(
        # never pull more often than every 1s.
        fastest=1,
        # never pull less often than every 1m.
        slowest=60,
        # start with 10s between iterations.
        start=10,
    )
    async for _ in hb:
        # Pull messages from a remote system.
        msgs = await pull_messages()
        # If there are more than 5 messages,
        # start pulling more often.
        if msgs > 5:
            hb.faster()
        # If there are no messages,
        # pull less often
        elif not msgs:
            hb.slower()
        ...  # do something with messages

🖨 感谢

该项目受cardio.go Go脚本启发，该脚本由Xe Iaso在其任何东西都可以成为消息队列，如果你使用它的方式足够错误博客文章中介绍。虽然实现和API不同，但为了使其符合Python风格和用户友好，进行了调整。