跳转到主要内容

使用anyio的AMQP实现

项目描述

Join chatroom Documentation Status Automated test status Test coverage

async_amqp

async_amqp 库是AMQP 0.9.1协议的纯Python实现。

基于anyio构建,它提供了一个基于协程的API,使得编写高度并发的应用程序变得容易。

欢迎提交错误报告、补丁和建议!只需打开问题或发送拉取请求

状态

代码可以正常工作。将使用aioamqp(甚至纯python-amqp)的代码移植过来应该是相当直接的。

所有来自aioamqp的测试都已移植并成功。

async_amqp 已从 asyncamqp 重命名为,但在pypi上已被采用。 asyncamqp 已从 trio_amqp 重命名,但anyio支持被认为是一个好主意。 trio_amqp 则是从 aioamqp 分叉而来。

测试

要运行测试,您需要安装Python测试依赖项

pip install -r ci/requirements_dev.txt

测试需要RabbitMQ的一个实例。您可以使用docker启动一个新的实例

docker run -d --log-driver=syslog -e RABBITMQ_NODENAME=my-rabbit --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

RabbitMQ需要一个具有管理员权限的“guest”用户(密码“guest”)。

您可以使用 make test 运行测试(需要 pytest)。

使用docker-compose进行测试

使用 docker-compose up -d rabbitmq 启动RabbitMQ。当RabbitMQ启动后,使用 docker-compose up --build aioamqp-test 启动测试

未来工作

  • 添加覆盖率报告。提高覆盖率(当然)。

  • 尝试重新启动失败的AMQP连接,而不是取消一切。

项目详情


下载文件

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

源分布

async_amqp-0.5.5.tar.gz (54.7 kB 查看哈希值)

上传时间

构建分布

async_amqp-0.5.5-py3-none-any.whl (21.2 kB 查看哈希值)

上传时间 Python 3

支持