Merge pull request #4 from Hanagotchi/HAN-18

HAN-18: Implementar conexión con RabbitMQ y recepción de paquetes

Author: violetaperezandrade

Dockerfile

@@ -7,6 +7,6 @@ RUN pip install -r requirements.txt
 
 EXPOSE 8080
 
-ADD app/ app/
+COPY app/ ./
 
-CMD ["uvicorn", "app.main:app", "--reload", "--host", "0.0.0.0", "--port", "8080"]
+CMD ["sh", "-c", "uvicorn main:app --reload --host 0.0.0.0 --port 8080 & python main_rabbitmq.py"]

app/common/__init__.py

@@ -4,8 +4,8 @@
 from os import environ
 from typing import Optional, Union
 
-from app.database.models.device_plant import DevicePlant
-from app.database.models.measurement import Measurement
+from database.models.device_plant import DevicePlant
+from database.models.measurement import Measurement
 
 load_dotenv()
 

app/database/database.py

@@ -1,7 +1,7 @@
 from sqlalchemy import Integer, String, SmallInteger
 from sqlalchemy.orm import Mapped, mapped_column
-from app.database.models.base import Base
-from app.schemas.device_plant import DevicePlantSchema
+from database.models.base import Base
+from schemas.device_plant import DevicePlantSchema
 
 
 class DevicePlant(Base):
@@ -14,13 +14,9 @@ class DevicePlant(Base):
     id_user: Mapped[int] = mapped_column(Integer)
 
     def __repr__(self) -> str:
-        return format(
-            "DevicePlant(id_device={0}, id_plant={1}, plant_type={2}, id_user={3})",
-            self.id_device,
-            self.id_plant,
-            self.plant_type,
-            self.id_user,
-        )
+        return (f"DevicePlant(id_device={self.id_device}, "
+                f"id_plant={self.id_plant}, "
+                f"plant_type={self.plant_type}, id_user={self.id_user})")
 
     @classmethod
     def from_pydantic(cls, pydantic_obj: DevicePlantSchema):

app/database/models/device_plant.py

@@ -1,7 +1,7 @@
 from sqlalchemy import CheckConstraint, Integer, String, SmallInteger
 from sqlalchemy.orm import Mapped, mapped_column
 from typing import Optional
-from app.database.models.base import Base
+from database.models.base import Base
 
 
 class Measurement(Base):

app/database/models/measurement.py

@@ -1,5 +1,5 @@
 from sqlalchemy import Column, Integer
-from app.database.models.base import Base
+from database.models.base import Base
 
 
 class Number(Base):

app/database/models/number.py

@@ -1,11 +1,11 @@
 from fastapi import FastAPI, Request, Body
-from app.database.database import SQLAlchemyClient
-from app.database.models.device_plant import DevicePlant
-from app.schemas.device_plant import DevicePlantSchema
+from database.database import SQLAlchemyClient
+from database.models.device_plant import DevicePlant
+from schemas.device_plant import DevicePlantSchema
 import logging
-from app.controller.calculator_controller import CalculatorController
-from app.schemas.schemas import Request as RequestSchema
-from app.service.calculator_service import CalculatorService
+from controller.calculator_controller import CalculatorController
+from schemas.schemas import Request as RequestSchema
+from service.calculator_service import CalculatorService
 
 app = FastAPI()
 service = CalculatorService()

app/main.py

@@ -0,0 +1,29 @@
+import os
+import logging
+from service.rabbitmq.consumer import Consumer
+
+
+def main():
+    logging_level = os.environ.get("LOGGING_LEVEL")
+    queue_name = os.environ.get("QUEUE_NAME")
+    initialize_log(logging_level)
+    consumer = Consumer(queue_name)
+    consumer.run()
+
+
+def initialize_log(logging_level):
+    """
+    Python custom logging initialization
+
+    Current timestamp is added to be able to identify in docker
+    compose logs the date when the log has arrived
+    """
+    logging.basicConfig(
+        format='%(asctime)s %(levelname)-8s %(message)s',
+        level=logging_level,
+        datefmt='%Y-%m-%d %H:%M:%S',
+    )
+
+
+if __name__ == "__main__":
+    main()

app/main_rabbitmq.py

@@ -1,7 +1,7 @@
 import os
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
-from app.database.models.number import Number
+from database.models.number import Number
 
 
 class CalculatorService:

app/service/calculator_service.py

@@ -0,0 +1,51 @@
+import pika
+import os
+
+
+class Middleware:
+
+    def __init__(self):
+        self._connection = (pika.BlockingConnection
+                            (pika.ConnectionParameters(host=os.environ.get
+                                                       ("RABBITMQ_HOST"))))
+        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/common/middleware.py

@@ -0,0 +1,3 @@
+FROM rabbitmq:3.12-management
+RUN apt-get update
+RUN apt-get install -y curl

app/service/rabbitmq/Dockerfile

@@ -0,0 +1,27 @@
+import json
+import logging
+from database.models.device_plant import DevicePlant
+from database.database import SQLAlchemyClient
+from ..common.middleware import Middleware
+
+
+class Consumer:
+    def __init__(self, queue_name):
+        self.__queue_name = queue_name
+        self.__middleware = Middleware()
+        self.__sqlAlchemyClient = SQLAlchemyClient()
+
+    def run(self):
+        self.__middleware.create_queue(self.__queue_name)
+        self.__middleware.listen_on(self.__queue_name, self.__callback)
+
+    def __callback(self, body):
+        body = json.loads(body)
+        print(body)
+        dp = DevicePlant(id_device=body["id_device"],
+                         id_plant=body["id_plant"],
+                         plant_type=body["plant_type"],
+                         id_user=body["id_user"])
+        self.__sqlAlchemyClient.add_new(dp)
+        logging.info("action: registro agregado a la base de datos|"
+                     f"device_plant: {dp}")

app/service/rabbitmq/__init__.py

@@ -0,0 +1,16 @@
+FROM postgres:15
+
+WORKDIR /
+
+ARG USE_POSTGIS=false
+
+RUN apt-get update \
+    && apt-get install -f -y --no-install-recommends \
+    ca-certificates \
+    software-properties-common \
+    build-essential \
+    pkg-config \
+    git \
+    postgresql-server-dev-15 \
+    && add-apt-repository "deb http://ftp.debian.org/debian testing main contrib" \
+    && apt-get update 

app/service/rabbitmq/consumer.py

@@ -1,5 +1,8 @@
-version: '3.8'
+version: '3.9'
+name: measurements
+
 services:
+
   python:
     build:
       context: .
@@ -9,19 +12,56 @@ services:
     ports:
       - "8080:8080"
     depends_on:
-      - sql
+      rabbitmq:
+        condition: service_healthy
+      sql:
+        condition: service_healthy
 
   rabbitmq:
-    image: rabbitmq:3.12-management
+    build:
+      context: ./app/service/rabbitmq
+      dockerfile: Dockerfile
     ports:
       - "5672:5672"
       - "15672:15672"
+    healthcheck:
+      test:
+        [
+          "CMD",
+          "rabbitmq-diagnostics",
+          "check_port_connectivity"
+        ]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+      start_period: 50s
 
   sql:
-    image: postgres:latest
+    build:
+      context: ./app/service/sql
+      dockerfile: Dockerfile
     env_file:
       - .env
     ports:
       - "5432:5432"
     volumes:
-      - ./sql:/docker-entrypoint-initdb.d
+      - ./app/service/sql:/docker-entrypoint-initdb.d
+    healthcheck:
+      test:
+        [
+          "CMD",
+          "pg_isready",
+          "-h",
+          "localhost",
+          "-p",
+          "5432",
+          "-q",
+          "-U",
+          "${POSTGRES_USER}",
+          "-d",
+          "${POSTGRES_DB}"
+        ]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+      start_period: 50s