Update environment variables and RabbitMQ connection

Author: fjpacheco

.env.dist

@@ -1,10 +1,12 @@
+
+LOGGING_LEVEL=
+
 POSTGRES_USER=
 POSTGRES_PASSWORD=
-MEASUREMENTS_DB=
 POSTGRES_HOST=
 POSTGRES_PORT=
-QUEUE_NAME=
-LOGGING_LEVEL=
-RABBITMQ_HOST=
-RABBITMQ_PORT=
-MEASUREMENTS_SCHEMA=
+MEASUREMENTS_DB=
+MEASUREMENTS_SCHEMA=
+
+RABBITMQ_URL=
+QUEUE_NAME=

app/database/models/device_plant.py

@@ -4,6 +4,7 @@
 from schemas.device_plant import DevicePlantSchema
 from os import environ
 
+
 class DevicePlant(Base):
     __tablename__ = "device_plant"
     __table_args__ = {'schema': environ.get("MEASUREMENTS_SCHEMA", "dev")}

app/database/models/measurement.py

@@ -4,6 +4,7 @@
 from database.models.base import Base
 from os import environ
 
+
 class Measurement(Base):
     __tablename__ = "measurements"
     __table_args__ = {'schema': environ.get("MEASUREMENTS_SCHEMA", "dev")}

app/service/calculator_service.py

@@ -1,4 +1,3 @@
-import os
 from sqlalchemy import create_engine, engine
 from sqlalchemy.orm import Session
 from database.models.number import Number

app/service/common/middleware.py

@@ -1,53 +1,51 @@
-import pika
-import os
-
-
-class Middleware:
-
-    def __init__(self):
-        rabbitmq_host = os.environ.get("RABBITMQ_HOST", "127.0.0.1")
-        rabbitmq_port = int(os.environ.get("RABBITMQ_PORT", 5672))
-        self._connection = pika.BlockingConnection(
-            pika.ConnectionParameters(host=rabbitmq_host, port=rabbitmq_port)
-        )
-        self._channel = self._connection.channel()
-        self._exit = False
-        self._remake = False
-
-    def create_queue(self, queue_name):
-        self._channel.queue_declare(queue=queue_name)
-
-    def _setup_message_consumption(self, queue_name, user_function):
-        self._channel.basic_consume(queue=queue_name,
-                                    on_message_callback=lambda channel,
-                                    method, properties, body:
-                                    (user_function(body),
-                                     channel.basic_ack
-                                     (delivery_tag=method.delivery_tag),
-                                     self._verify_connection_end()))
-        self._channel.start_consuming()
-
-    def _verify_connection_end(self):
-        if self._exit:
-            self._channel.close()
-            if self._remake:
-                self._exit = False
-                self._channel = self._connection.channel()
-
-    def finish(self, open_new_channel=False):
-        self._exit = True
-        self._remake = open_new_channel
-
-    # Work queue methods
-    def listen_on(self, queue_name, user_function):
-        self.create_queue(queue_name)
-        self._channel.basic_qos(prefetch_count=30)
-        self._setup_message_consumption(queue_name, user_function)
-
-    def send_message(self, queue_name, message):
-        self._channel.basic_publish(exchange='',
-                                    routing_key=queue_name,
-                                    body=message)
-
-    def __del__(self):
-        self._connection.close()
+import pika
+import os
+
+
+class Middleware:
+
+    def __init__(self):
+        self._connection = pika.BlockingConnection(
+            parameters=pika.URLParameters(
+                os.environ.get("RABBITMQ_URL", "amqp://rabbitmq:5672")))
+        self._channel = self._connection.channel()
+        self._exit = False
+        self._remake = False
+
+    def create_queue(self, queue_name):
+        self._channel.queue_declare(queue=queue_name)
+
+    def _setup_message_consumption(self, queue_name, user_function):
+        self._channel.basic_consume(queue=queue_name,
+                                    on_message_callback=lambda channel,
+                                    method, properties, body:
+                                    (user_function(body),
+                                     channel.basic_ack
+                                     (delivery_tag=method.delivery_tag),
+                                     self._verify_connection_end()))
+        self._channel.start_consuming()
+
+    def _verify_connection_end(self):
+        if self._exit:
+            self._channel.close()
+            if self._remake:
+                self._exit = False
+                self._channel = self._connection.channel()
+
+    def finish(self, open_new_channel=False):
+        self._exit = True
+        self._remake = open_new_channel
+
+    # Work queue methods
+    def listen_on(self, queue_name, user_function):
+        self.create_queue(queue_name)
+        self._channel.basic_qos(prefetch_count=30)
+        self._setup_message_consumption(queue_name, user_function)
+
+    def send_message(self, queue_name, message):
+        self._channel.basic_publish(exchange='',
+                                    routing_key=queue_name,
+                                    body=message)
+
+    def __del__(self):
+        self._connection.close()

app/service/rabbitmq/consumer.py

@@ -1,4 +1,3 @@
-import os
 import json
 from exceptions.logger_messages import LoggerMessages
 import pydantic