diff --git a/amqp/rabbitmq/rabbit_mq_client.py b/amqp/rabbitmq/rabbit_mq_client.py index 0c3ca58..517b422 100644 --- a/amqp/rabbitmq/rabbit_mq_client.py +++ b/amqp/rabbitmq/rabbit_mq_client.py @@ -1,8 +1,7 @@ import logging import aio_pika -from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange -from pika.exchange_type import ExchangeType +from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange, ExchangeType from amqp.adapter.data_message_factory import DataMessageFactory from amqp.adapter.logging_utils import ( @@ -55,7 +54,7 @@ class RabbitMQClient: name=cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False ) self.reply_to_exchange = await self.channel.declare_exchange( - AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.direct + AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.DIRECT ) await self.reply_to_queue.bind( exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue @@ -114,7 +113,7 @@ class RabbitMQClient: ) exchange: AbstractRobustExchange = ( await self.channel.declare_exchange( - name=_exchange_name, type=ExchangeType.topic + name=_exchange_name, type=ExchangeType.TOPIC ) ) await queue.bind(exchange, route.key) diff --git a/amqp/router/router_producer.py b/amqp/router/router_producer.py index 174b502..e78c2ec 100644 --- a/amqp/router/router_producer.py +++ b/amqp/router/router_producer.py @@ -15,9 +15,8 @@ from aio_pika.abc import ( AbstractRobustQueue, AbstractRobustConnection, AbstractRobustExchange, - AbstractIncomingMessage, + AbstractIncomingMessage, ExchangeType, ) -from pika.exchange_type import ExchangeType from amqp.adapter.logging_utils import ( logging_error, @@ -82,13 +81,13 @@ class RouterProducer(RouterBase): # **** set up the service discovery internal exchanges **** # 1. declare RequestExchange to where to publish the RoutingData requests self.cm_request_exchange = await self.channel.declare_exchange( - name=CM_REQUEST_EXCHANGE, type=ExchangeType.fanout + name=CM_REQUEST_EXCHANGE, type=ExchangeType.FANOUT ) # # 2. declare ResponseExchange where to publish current RoutingData (response to # a RoutingData request) self.cm_response_exchange = await self.channel.declare_exchange( - name=CM_RESPONSE_EXCHANGE, type=ExchangeType.fanout + name=CM_RESPONSE_EXCHANGE, type=ExchangeType.FANOUT ) # 2a. declare a Response queue for Routing Data replies from Services _cm_response_queue: str = self.get_response_queue_name() @@ -145,9 +144,9 @@ class RouterProducer(RouterBase): cm_message.reply_to, ) if ( - cm_message.is_valid() - and not self.amq_configuration.amq_adapter.service_name - == cm_message.reply_to + cm_message.is_valid() + and not self.amq_configuration.amq_adapter.service_name + == cm_message.reply_to ): if cm_message.message_type == ServiceMessageType.ROUTING_DATA: self.add_routes(cm_message.message) @@ -167,7 +166,7 @@ class RouterProducer(RouterBase): ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT ) if not await self.publish_service_message( - serviceMessage, self.cm_request_exchange + serviceMessage, self.cm_request_exchange ): logging_warning( "RouterProducer could not request route data: %s channel is not open.", @@ -178,7 +177,7 @@ class RouterProducer(RouterBase): logging_error("RouterProducer failed request routing data: %s", ioe) async def publish_service_message( - self, message: ServiceMessage, exchange: AbstractRobustExchange + self, message: ServiceMessage, exchange: AbstractRobustExchange ) -> bool: """ * Publishes a service message to the service queue and logs success log entry. diff --git a/pyproject.toml b/pyproject.toml index d103962..ed8e772 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,10 +19,10 @@ classifiers = [ ] requires-python = ">=3.11" dependencies = [ - "aio-pika~=9.5.5", - "aiohttp~=3.11.11", - "aio_pika~=9.5.5", - "fastapi==0.114.2", + "aio-pika~=9.5", + "aiohttp~=3.11", + "aio_pika~=9.5", + "fastapi~=0.114", "opentelemetry-api", "opentelemetry-sdk", "opentelemetry-exporter-otlp",