Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

RED-4653: Removed variable re-use to make the code clearer

Browse Source
This commit is contained in:
Viktor Seifert 2022-07-22 14:11:48 +02:00
parent 4bf1a02345
commit 379685f964
1 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
pyinfra/queue/queue_manager.py
View File

@ -8,11 +8,8 @@ from pyinfra.config import get_config, Config
CONFIG = get_config()
logger = logging.getLogger("pika")
logger.setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
logger.setLevel(CONFIG.logging_level_root)
pika_logger = logging.getLogger("pika")
pika_logger.setLevel(logging.WARNING)
def get_connection_params(config: Config) -> pika.ConnectionParameters:
@ -49,40 +46,43 @@ class QueueManager(object):
self._consumer_token = None
self.logger = logging.getLogger("queue_manager")
self.logger.setLevel(CONFIG.logging_level_root)
def start_consuming(self, process_message_callback: Callable):
callback = self._create_queue_callback(process_message_callback)
logger.info("Consuming from queue")
self.logger.info("Consuming from queue")
consumer_tag = None
try:
consumer_tag = self._channel.basic_consume(self._input_queue, callback)
logger.info(f"Registered with consumer-tag: {consumer_tag}")
self.logger.info(f"Registered with consumer-tag: {consumer_tag}")
self._channel.start_consuming()
finally:
logger.warning("An unhandled exception occurred while consuming messages. Consuming will stop.")
self.logger.warning("An unhandled exception occurred while consuming messages. Consuming will stop.")
if consumer_tag:
logger.info(f"Cancelling subscription for consumer-tag: {consumer_tag}")
self.logger.info(f"Cancelling subscription for consumer-tag: {consumer_tag}")
self._channel.basic_cancel(consumer_tag)
else:
logger.warning("No consumer-tag found, a possible subscription cannot be cancelled")
self.logger.warning("No consumer-tag found, a possible subscription cannot be cancelled")
def _create_queue_callback(self, process_message_callback: Callable):
def callback(_channel, frame, properties, body):
logger.info("Received message from queue")
logger.debug(f"Processing {(frame, properties, body)}.")
self.logger.info("Received message from queue")
self.logger.debug(f"Processing {(frame, properties, body)}.")
try:
unpacked_message_body = json.loads(body)
callback_result = process_message_callback(unpacked_message_body)
logger.info("Processed message, publishing result to result-queue")
self.logger.info("Processed message, publishing result to result-queue")
self._channel.basic_publish("", self._output_queue, json.dumps(callback_result).encode())
self._channel.basic_ack(frame.delivery_tag)
except Exception as ex:
n_attempts = _get_n_previous_attempts(properties) + 1
logger.warning(f"Failed to process message, {n_attempts} attempts, error: {str(ex)}")
self.logger.warning(f"Failed to process message, {n_attempts} attempts, error: {str(ex)}")
raise ex
return callback