Merge pull request #5 from MissterHao/feature/logging

Add logger module and use it in dysession

.coveragerc

@@ -2,6 +2,8 @@
 omit =
     tests/*.py
     runtests.py
+    dysession/logger/handler/colorful_console.py
+
 [report]
 # Regexes for lines to exclude from consideration
 exclude_lines =

dysession/aws/dynamodb.py

@@ -13,6 +13,7 @@
     SessionKeyDuplicated,
 )
 from dysession.backends.model import SessionDataModel
+from dysession.logger import get_logger
 
 from ..settings import get_config
 
@@ -58,7 +59,7 @@ def check_dynamodb_table_exists(table_name: Optional[str] = None, client=None) -
 
     response = client.list_tables()
     if table_name not in response["TableNames"]:
-        raise DynamodbTableNotFound
+        raise DynamodbTableNotFound(table_name)
 
     return response
 
@@ -128,6 +129,8 @@ def insert_session_item(
         table_name = get_config()["DYNAMODB_TABLENAME"]
 
     if not ignore_duplicated and key_exists(data.session_key):
+        logger = get_logger()
+        logger.error(f"'{data.session_key}' is already an item of table '{table_name}'.")
         raise SessionKeyDuplicated
 
     resource = boto3.resource("dynamodb", region_name=get_config()["DYNAMODB_REGION"])
@@ -200,9 +203,13 @@ def get(
                     raise SessionExpired
         # if not found then raise
         except DynamodbItemNotFound:
+            logger = get_logger()
+            logger.error(f"'{session_key}' cannot be found on table '{table_name}'.")
             raise SessionKeyDoesNotExist
         # if key is expired
         except SessionExpired:
+            logger = get_logger()
+            logger.error(f"'{session_key}' is expired .")
             raise SessionExpired
 
         return model
@@ -236,7 +243,7 @@ def exists(self, session_key: str) -> bool:
     def delete(self, data: SessionDataModel, table_name: Optional[str] = None) -> bool:
         if data.session_key is None:
             return
-            
+
         try:
             delete_session_item(data=data, table_name=table_name)
         except AssertionError:

dysession/aws/error.py

@@ -1,6 +1,16 @@
+from typing import Optional
+
+from dysession.logger import get_logger
+
 
 class DynamodbTableNotFound(Exception):
-    pass
+    def __init__(self, table_name: Optional[str] = None, *args: object) -> None:
+        super().__init__(*args)
+
+        logger = get_logger()
+        if table_name:
+            logger.error(f"'{table_name}' is not found in current region.")
+
 
 class DynamodbItemNotFound(Exception):
-    pass
+    pass

dysession/logger/__init__.py

@@ -0,0 +1,71 @@
+import logging
+import sys
+from enum import Enum, auto
+from functools import lru_cache
+from typing import Literal
+
+from .handler.colorful_console import ColorfulConsoleLoggerHandler
+
+
+class LoggingType(Enum):
+
+    PLAINTEXT_CONSOLE = auto()
+    COLOR_CONSOLE = auto()
+    FILE = auto()
+    CONSOLE = auto()
+
+
+@lru_cache
+def is_tty() -> bool:
+    """Hepler function with lru_cache"""
+    return sys.stdout.isatty()
+
+
+def get_logger(
+    logger_name: str = "dysession",
+    logger_type: Literal[LoggingType.CONSOLE, LoggingType.FILE] = LoggingType.CONSOLE,
+    level: int = logging.DEBUG,
+) -> logging.Logger:
+    """
+    This function return a logging.Logger with handlers.
+    Handlers could be `ColorfulConsoleLoggerHandler`, `StreamHandler`, `FileHandler`.
+
+    ```
+    logger = get_logger()
+    logger.debug("This is a DEBUG log.")
+    logger.info("This is a INFO log.")
+    logger.warning("This is a WARNING log.")
+    logger.critical("This is a CRITICAL log.")
+    logger.fatal("This is a FATAL log.")
+    ```
+    """
+
+    logger = logging.getLogger(logger_name)
+    format = logging.Formatter(
+        "[%(asctime)-s] [%(levelname)-8s] %(name)s %(message)s ... ( %(filename)s:%(levelno)s )"
+    )
+    logger.setLevel(level)
+
+    if not logger.handlers:
+        if logger_type == LoggingType.CONSOLE:
+            if is_tty():
+                handler = ColorfulConsoleLoggerHandler()
+            else:
+                handler = logging.StreamHandler()
+        elif logger_type == LoggingType.FILE:
+            handler = logging.FileHandler("session.log", "a", encoding="utf-8")
+
+        handler.setFormatter(format)
+        logger.addHandler(handler)
+    else:
+        logger.warning(
+            f"Dysession logger had already initialized with logger({logger.handlers})"
+        )
+
+    return logger
+
+
+__all__ = (
+    LoggingType,
+    get_logger,
+)

dysession/logger/handler/ansi.py

@@ -0,0 +1,19 @@
+from enum import Enum
+
+
+class ANSIColor(Enum):
+    HEADER = "\033[95m"
+    OKBLUE = "\033[94m"
+    OKCYAN = "\033[96m"
+    OKGREEN = "\033[92m"
+    SUCCESSFUL = "\033[38;5;107m"  # 7b9246
+    WARNING = "\033[93m"
+    FAIL = "\033[91m"
+    DEBUG = "\033[37m"
+    ENDC = "\033[0m"
+    BOLD = "\033[1m"
+    UNDERLINE = "\033[4m"
+
+
+def colorful_it(color: ANSIColor, content: str) -> str:
+    return f"{color.value}{content}{ANSIColor.ENDC.value}"

dysession/logger/handler/colorful_console.py

@@ -0,0 +1,36 @@
+import logging
+
+from .ansi import ANSIColor
+
+LOGLEVEL_TRANSFORM = {
+    logging.DEBUG: ANSIColor.DEBUG.value,
+    logging.INFO: ANSIColor.OKCYAN.value,
+    logging.WARNING: ANSIColor.WARNING.value,
+    logging.ERROR: ANSIColor.FAIL.value,
+    logging.CRITICAL: ANSIColor.FAIL.value,
+    logging.FATAL: ANSIColor.FAIL.value,
+}
+
+
+class ColorfulConsoleLoggerHandler(logging.StreamHandler):
+    """
+    A handler class which allows the cursor to stay on
+    one line for selected messages
+    """
+
+    def emit(self, record):
+        try:
+            record.levelname = (
+                LOGLEVEL_TRANSFORM[record.levelno]
+                + f"{record.levelname:>8}"
+                + ANSIColor.ENDC.value
+            )
+
+            msg = self.format(record)
+            self.stream.write(msg)
+            self.stream.write(self.terminator)
+            self.flush()
+        except (KeyboardInterrupt, SystemExit):
+            raise
+        except:
+            self.handleError(record)

dysession/management/commands/dysession_init.py

@@ -5,6 +5,7 @@
 from django.core.management.base import BaseCommand, CommandParser
 
 from dysession.aws.dynamodb import create_dynamodb_table
+from dysession.logger import get_logger
 from dysession.settings import get_config
 
 from ._arg_types import positive_int
@@ -64,4 +65,7 @@ def add_arguments(self, parser: CommandParser) -> None:
         return super().add_arguments(parser)
 
     def handle(self, *args: Any, **options: Any) -> Optional[str]:
+        logger = get_logger()
+        logger.info("Start command: initialize dynamodb table")
         create_dynamodb_table(options=options)
+        logger.info("End of command: dynamodb table created successfully!")

dysession/middleware.py

@@ -7,6 +7,4 @@ def process_request(self, request):
         # SESSION_COOKIE_NAME can be change by developers
         # https://docs.djangoproject.com/en/3.2/ref/settings/#session-cookie-name
         session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME)
-        request.session = self.SessionStore(
-            session_key=session_key,
-        )
+        request.session = self.SessionStore(session_key=session_key)

dysession/version.txt

@@ -1 +1 @@
-1.0.3
+1.1.0

tests/test_aws_dynamodb.py

@@ -1,3 +1,5 @@
+import logging
+
 import boto3
 from django.test import TestCase
 from moto import mock_dynamodb
@@ -18,6 +20,12 @@
 
 
 class AWSDynamoDBTestCase(TestCase):
+    def setUp(self):
+        logging.disable(logging.CRITICAL)
+
+    def tearDown(self):
+        logging.disable(logging.NOTSET)
+
     @mock_dynamodb
     def test_init_dynamodb_table(self):
 

tests/test_backend_db.py

@@ -1,3 +1,4 @@
+import logging
 import time
 from datetime import datetime
 from typing import Any
@@ -11,10 +12,8 @@
     DynamoDB,
     check_dynamodb_table_exists,
     create_dynamodb_table,
-    destory_dynamodb_table,
     get_item,
     insert_session_item,
-    key_exists,
 )
 from dysession.aws.error import DynamodbItemNotFound, DynamodbTableNotFound
 from dysession.backends.error import (
@@ -27,6 +26,12 @@
 
 
 class DynamoDBTestCase(TestCase):
+    def setUp(self):
+        logging.disable(logging.CRITICAL)
+
+    def tearDown(self):
+        logging.disable(logging.NOTSET)
+
     @mock_dynamodb
     def create_dynamodb_table(self):
         self.options = {
@@ -266,7 +271,6 @@ def test_delete_item_via_dynamodb_controller_raise_error(self):
         with self.assertRaises(AssertionError):
             db.delete(model)
 
-
     @mock_dynamodb
     def test_delete_item_via_dynamodb_controller_with_none_type_session_key(self):
 
@@ -279,4 +283,4 @@ def test_delete_item_via_dynamodb_controller_with_none_type_session_key(self):
         model.session_key = None
 
         db = DynamoDB(self.client)
-        db.delete(model)
+        db.delete(model)

tests/test_logger.py

@@ -0,0 +1,61 @@
+import logging
+import sys
+from unittest import mock
+
+from django.test import TestCase
+from parameterized import parameterized
+
+from dysession.logger import LoggingType, get_logger, is_tty
+from dysession.logger.handler.colorful_console import ColorfulConsoleLoggerHandler
+
+
+class LoggerTestCase(TestCase):
+    def test_get_logger_with_second_logger_type_passed_in(self):
+
+        with self.assertLogs("test_get_logger_with_second_logger_type_passed_in") as cm:
+            logger = get_logger("test_get_logger_with_second_logger_type_passed_in")
+            logger.info("This is a test content")
+        self.assertIn("This is a test content", "\n".join(cm.output))
+
+        with self.assertLogs("test_get_logger_with_second_logger_type_passed_in") as cm:
+            logger = get_logger("test_get_logger_with_second_logger_type_passed_in", logger_type=LoggingType.FILE)
+            logger.info("This is a test content")
+        self.assertIn("WARNING", "\n".join(cm.output))
+
+    @mock.patch("sys.stdout.isatty")
+    def test_logger_is_tty_handlers(self, mock_is_tty):
+        is_tty.cache_clear()
+        mock_is_tty.return_value = True
+
+        logger = get_logger("test_logger_is_tty_handlers")
+        self.assertEqual(len(logger.handlers), 1)
+        self.assertEqual(logger.name, "test_logger_is_tty_handlers")
+        self.assertIs(type(logger.handlers[0]), ColorfulConsoleLoggerHandler)
+
+        logger.handlers = []
+
+    @mock.patch("sys.stdout.isatty")
+    def test_logger_is_not_tty_handlers(self, mock_is_tty):
+        is_tty.cache_clear()
+        mock_is_tty.return_value = False
+
+        logger = get_logger("test_logger_is_not_tty_handlers")
+        self.assertEqual(len(logger.handlers), 1)
+        self.assertEqual(logger.name, "test_logger_is_not_tty_handlers")
+        self.assertIs(type(logger.handlers[0]), logging.StreamHandler)
+
+        logger.handlers = []
+
+    def test_logger_file_handler(self):
+
+        logger = get_logger("test_logger_file_handler", logger_type=LoggingType.FILE)
+
+        self.assertEqual(len(logger.handlers), 1)
+        self.assertEqual(logger.name, "test_logger_file_handler")
+        self.assertIsInstance(logger.handlers[0], logging.FileHandler)
+
+        with self.assertLogs("test_logger_file_handler") as cm:
+            logger.info("This is a test content")
+        self.assertIn("This is a test content", "\n".join(cm.output))
+
+        logger.handlers = []

tests/test_logger_handler_ansi.py

@@ -0,0 +1,30 @@
+from typing import Any
+
+from django.test import TestCase
+from parameterized import parameterized
+
+from dysession.logger import get_logger
+from dysession.logger.handler.ansi import ANSIColor, colorful_it
+
+
+class ANSITestCase(TestCase):
+    @parameterized.expand(
+        [
+            [ANSIColor.HEADER],
+            [ANSIColor.OKBLUE],
+            [ANSIColor.OKCYAN],
+            [ANSIColor.OKGREEN],
+            [ANSIColor.SUCCESSFUL],
+            [ANSIColor.WARNING],
+            [ANSIColor.FAIL],
+            [ANSIColor.DEBUG],
+            [ANSIColor.BOLD],
+            [ANSIColor.UNDERLINE],
+        ]
+    )
+    def test_ansi_colorful_it_func(self, ansi_color: ANSIColor):
+
+        self.assertEqual(
+            colorful_it(ansi_color, "Content"),
+            f"{ansi_color.value}Content{ANSIColor.ENDC.value}",
+        )