Merge pull request #24 from meedan/devops-471-logging-and-http

DEVOPS-471 add logging and simultaneously run HTTP server

Makefile

@@ -1,6 +1,9 @@
 .PHONY: run run_http run_worker run_test
 
-run: $(if $(filter http,$(RUN_MODE)),run_http,run_worker)
+run: 
+	uvicorn main:app --reload &
+	python run.py &
+	wait
 
 run_http:
 	uvicorn main:app --reload

lib/http.py

@@ -3,11 +3,20 @@
 from typing import Any, Dict
 import httpx
 from httpx import HTTPStatusError
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from pydantic import BaseModel
 from lib.queue.queue import Queue
+from lib.logger import logger
 
 app = FastAPI()
+
+@app.middleware("http")
+async def log_requests(request: Request, call_next):
+    logger.info(f"Calling endpoint: {request.url.path}")
+    response = await call_next(request)
+    logger.info(f"Endpoint {request.url.path} returned: {response.status_code}")
+    return response
+
 async def post_url(url: str, params: dict) -> Dict[str, Any]:
     async with httpx.AsyncClient() as client:
         try:

lib/logger.py

@@ -0,0 +1,38 @@
+import logging
+import logging.config
+
+LOGGING_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "formatters": {
+        "default": {
+            "format": "{asctime} - {levelname} - {name} - {message}",
+            "style": "{",
+        }
+    },
+    "handlers": {
+        "console": {
+            "class": "logging.StreamHandler",
+            "formatter": "default",
+        }
+    },
+    "loggers": {
+        "uvicorn": {
+            "handlers": ["console"],
+            "level": "INFO",
+        },
+        "uvicorn.access": {
+            "handlers": ["console"],
+            "level": "INFO",
+        },
+    },
+    "root": {
+        "handlers": ["console"],
+        "level": "INFO",
+    }
+}
+
+logging.config.dictConfig(LOGGING_CONFIG)
+
+# This provides an easily accessible logger for other modules
+logger = logging.getLogger(__name__)

lib/queue/queue.py

@@ -5,6 +5,7 @@
 
 from lib.helpers import get_class, get_setting
 from lib.model.model import Model
+from lib.logger import logger
 
 class Queue(ABC):
     @classmethod
@@ -13,6 +14,7 @@ def create(cls, input_queue_name: str = None, output_queue_name: str = None, que
         Instantiate a queue. Must pass queue_driver_name (i.e. sqs_queue.SQSQueue vs redis_queue.RedisQueue), 
         input_queue_name, output_queue_name, and batch_size. Pulls settings and then inits instance.
         """
+        logger.info(f"Starting queue with: ({input_queue_name}, {output_queue_name}, {queue_driver_name}, {batch_size})")
         input_queue_name = get_setting(input_queue_name, "INPUT_QUEUE_NAME")
         output_queue_name = get_setting(output_queue_name, "OUTPUT_QUEUE_NAME")
         return get_class('lib.queue.', get_setting(queue_driver_name, "QUEUE_TYPE"))(input_queue_name, output_queue_name, batch_size)
@@ -38,10 +40,10 @@ def safely_respond(self, model: Model) -> Tuple[List[Dict[str, str]], List[Dict[
         Return responses if no failure.
         """
         messages = self.receive_messages(model.BATCH_SIZE)
-        try:
+        responses = []
+        if messages:
+            logger.info(f"About to respond to: ({messages})")
             responses = model.respond(copy.deepcopy(messages))
-        except:
-            responses = []
         return messages, responses
 
     def fingerprint(self, model: Model):
@@ -53,6 +55,7 @@ def fingerprint(self, model: Model):
         messages, responses = self.safely_respond(model)
         if responses:
             for message, response in zip(messages, responses):
+                logger.info(f"Processing message of: ({message}, {response})")
                 try:
                     self.return_response({"request": message, "response": response})
                 except:

test/lib/queue/test_queue.py

@@ -26,7 +26,7 @@ def test_create(self):
             Queue.create('input', 'output', 'invalidqueue', 2)
 
     def test_fingerprint(self):
-        self.queue.receive_messages = MagicMock(return_value=[schemas.TextInput(id="123", callback_url="http://example.com?callback=1", text="msg1")])
+        self.queue.receive_messages = MagicMock(return_value=[schemas.Message(body=schemas.TextInput(id="123", callback_url="http://example.com?callback=1", text="msg1"))])
         self.queue.input_queue = MagicMock(return_value=None)
         self.model.model = self.mock_model
         self.model.model.encode = MagicMock(return_value=np.array([[4, 5, 6], [7, 8, 9]]))