aiomisc - asyncio的杂项工具
项目描述
为asyncio提供的杂项工具。
作为程序员，您在构建和维护软件应用程序时面临的挑战并不陌生。其中特别困难的一个领域是构建使用异步I/O的软件架构。
这里是aiomisc发挥作用的地方。aiomisc是一个Python库，它提供了一组工具函数和类，用于更直观和高效地处理异步I/O。它基于asyncio库构建，旨在帮助开发者编写更可靠和可扩展的异步代码。
使用aiomisc，您可以利用如工作池、连接池、断路器模式以及asyncbackoff和asyncretry等强大的功能，使您的asyncio代码更加健壮并易于维护。在本文档中，我们将深入了解aiomisc提供的功能以及它是如何帮助您简化异步服务开发的。
安装
安装可以通过标准方式完成，例如通过PyPI或直接从git仓库安装。
从PyPI安装
pip3 install aiomisc从github.com安装
# Using git tool
pip3 install git+https://github.com/aiokitchen/aiomisc.git
# Alternative way using http
pip3 install \
https://github.com/aiokitchen/aiomisc/archive/refs/heads/master.zip该软件包包含一些额外功能，并且您可以通过指定它们来安装额外的依赖项。
使用uvloop
pip3 install "aiomisc[uvloop]"使用aiohttp
pip3 install "aiomisc[aiohttp]"以下为完整额外功能表
示例 |
描述 |
|---|---|
pip install aiomisc[aiohttp] |
用于运行aiohttp应用程序。 |
pip install aiomisc[asgi] |
用于运行ASGI应用程序。 |
pip install aiomisc[carbon] |
|
pip install aiomisc[cron] |
使用croniter进行任务调度。 |
pip install aiomisc[raven] |
|
pip install aiomisc[rich] |
您可能需要使用rich进行日志记录。 |
pip install aiomisc[uvicorn] |
|
pip install aiomisc[uvloop] |
使用uvloop作为默认的事件循环。 |
您可以通过逗号分隔值来组合额外功能，例如
pip3 install "aiomisc[aiohttp,cron,rich,uvloop]"快速入门
本节将介绍如何创建和使用事件循环以及创建服务。当然，这里不能涵盖所有内容，但您可以在教程部分阅读很多内容，并且您始终可以参考模块和API参考部分以获取帮助。
事件循环和入口点
让我们先看这个简单的例子
import asyncio
import logging
import aiomisc
log = logging.getLogger(__name__)
async def main():
log.info('Starting')
await asyncio.sleep(3)
log.info('Exiting')
if __name__ == '__main__':
with aiomisc.entrypoint(log_level="info", log_format="color") as loop:
loop.run_until_complete(main())此代码声明了一个异步main()函数，它在3秒后退出。这似乎没有什么有趣的，但关键点在于entrypoint。
entrypoint做了什么，看起来并不是很多，它创建了一个事件循环并将控制权转移给用户。然而，在底层，日志记录器在单独的线程中进行配置，创建了一个线程池，启动了服务，但更多内容将在后面介绍，在这个例子中没有服务。
或者，您可以选择不使用入口点，只需创建一个事件循环并将其设置为当前线程的默认事件循环
import asyncio
import aiomisc
# * Installs uvloop event loop is it's has been installed.
# * Creates and set `aiomisc.thread_pool.ThreadPoolExecutor`
# as a default executor
# * Sets just created event-loop as a current event-loop for this thread.
aiomisc.new_event_loop()
async def main():
await asyncio.sleep(1)
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(main())上述示例在您的代码已经使用隐式创建的事件循环时非常有用，您需要修改的代码更少，只需添加aiomisc.new_event_loop()，并且所有对asyncio.get_event_loop()的调用都将返回创建的实例。
但是，您可以通过一次调用完成操作。以下示例隐式关闭了 asyncio 事件循环并安装了一个新的循环
import asyncio
import aiomisc
async def main():
await asyncio.sleep(3)
if __name__ == '__main__':
loop = aiomisc.new_event_loop()
loop.run_until_complete(main())服务
入口点的最主要功能是启动和优雅地停止服务。
在这个库中，服务概念指的是从 aiosmic.Service 类派生的类，并实现了 async def start(self) -> None: 方法，以及可选的 async def stop(self, exc: Optional[ Exception]) -> None 方法。
停止服务的概念不一定是指用户按下 Ctrl+C 键，实际上只是退出 entrypoint 的上下文管理器。
以下示例显示了您的服务可能的样子
from aiomisc import entrypoint, Service
class MyService(Service):
async def start(self):
do_something_when_start()
async def stop(self, exc):
do_graceful_shutdown()
with entrypoint(MyService()) as loop:
loop.run_forever()入口点可以启动任意数量的服务实例，并且它们将并发启动。
如果 start 方法是服务的有效负载，那么不需要实现停止方法，因为运行带有 start 函数的任务将在停止阶段被取消。但是在这种情况下，您必须通知 entrypoint 服务实例的初始化已完成，它可以继续。
如下所示
import asyncio
from threading import Event
from aiomisc import entrypoint, Service
event = Event()
class MyService(Service):
async def start(self):
# Send signal to entrypoint for continue running
self.start_event.set()
await asyncio.sleep(3600)
with entrypoint(MyService()) as loop:
assert event.is_set()这个库的全部威力在于已经实现或抽象的服务集。例如：AIOHTTPService，ASGIService，TCPServer，UDPServer，TCPClient，PeriodicService，CronService 等等。
不幸的是，在这一节中无法对此给予更多关注，请关注教程部分，那里有更多示例和解释，当然您也可以在 /api/index 或源代码中找到答案。作者已经尽力使源代码尽可能清晰和简单，所以您可以自由地探索它。
版本控制
本软件遵循 语义版本控制
总结：给出了版本号 MAJOR.MINOR.PATCH，当您进行不兼容的 API 变更时增加
MAJOR 版本
MINOR 版本当您以向后兼容的方式添加功能时
PATCH 版本当您进行向后兼容的错误修复时
可用作 MAJOR.MINOR.PATCH 格式扩展的预发布标签和构建元数据标签。
在这种情况下，包版本由 poem-plugins 自动分配，它使用存储库中的标签作为主版本和次版本以及计数器，该计数器是标签和分支头之间的提交数。
总结：给出了版本号 MAJOR.MINOR.PATCH，当您进行不兼容的 API 变更时增加
MAJOR 版本
MINOR 版本当您以向后兼容的方式添加功能时
PATCH 版本当您进行向后兼容的错误修复时
可用作 MAJOR.MINOR.PATCH 格式扩展的预发布标签和构建元数据标签。
在这种情况下，包版本由 poem-plugins 自动分配，它使用存储库中的标签作为主版本和次版本以及计数器，该计数器是标签和分支头之间的提交数。
如何开发？
像大多数开源项目一样，该项目由爱好者开发，您可以加入开发、提交问题或发送您的合并请求。
要开始在本存储库中开发，您需要执行以下操作。
应安装
Python 3.7+ 作为 python3
安装 Poetry 作为 poetry
为了设置开发环境，只需执行
# installing all dependencies poetry install # setting up pre-commit hooks poetry run pre-commit install # adding poem-plugins to the poetry poetry self add poem-plugins
下载文件
下载适合您平台文件的文件。如果您不确定选择哪一个，请了解更多关于安装包的信息。