First commit - Adding Buffer and Config classes

Author: leandrodamascena

aws_lambda_powertools/logging/buffer/__init__.py

@@ -0,0 +1,4 @@
+from aws_lambda_powertools.logging.buffer.cache import LoggerBufferCache
+from aws_lambda_powertools.logging.buffer.config import LoggerBufferConfig
+
+__all__ = ["LoggerBufferCache", "LoggerBufferConfig"]

aws_lambda_powertools/logging/buffer/cache.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import warnings
+from collections import deque
+from typing import Any
+
+from aws_lambda_powertools.warnings import PowertoolsUserWarning
+
+
+class LoggerBufferCache:
+    def __init__(self, max_size_bytes: int):
+        """
+        Initialize the LoggerBufferCache.
+
+        Parameters
+        ----------
+        max_size_bytes : int
+            Maximum size of the cache in bytes.
+        """
+        self.max_size_bytes: int = max_size_bytes
+        self.cache: dict[str, deque] = {}
+        self.current_size: dict[str, int] = {}
+
+    def add(self, key: str, item: Any) -> None:
+        """
+        Add an item to the cache for a specific key.
+
+        Parameters
+        ----------
+        key : str
+            The key to store the item under.
+        item : Any
+            The item to be stored in the cache.
+
+        Notes
+        -----
+        If the item size exceeds the maximum cache size, it will not be added.
+        """
+        item_size = len(str(item))
+
+        if item_size > self.max_size_bytes:
+            warnings.warn(
+                message=f"Item size {item_size} bytes exceeds total cache size {self.max_size_bytes} bytes",
+                category=PowertoolsUserWarning,
+                stacklevel=2,
+            )
+            return
+
+        if key not in self.cache:
+            self.cache[key] = deque()
+            self.current_size[key] = 0
+
+        while self.current_size[key] + item_size > self.max_size_bytes and self.cache[key]:
+            removed_item = self.cache[key].popleft()
+            self.current_size[key] -= len(str(removed_item))
+
+        self.cache[key].append(item)
+        self.current_size[key] += item_size
+
+    def get(self, key: str) -> list:
+        """
+        Retrieve items for a specific key.
+
+        Parameters
+        ----------
+        key : str
+            The key to retrieve items for.
+
+        Returns
+        -------
+        list
+            List of items for the given key, or an empty list if the key doesn't exist.
+        """
+        return list(self.cache.get(key, deque()))
+
+    def clear(self, key: str | None = None) -> None:
+        """
+        Clear the cache, either for a specific key or entirely.
+
+        Parameters
+        ----------
+        key : str, optional
+            The key to clear. If None, clears the entire cache.
+        """
+        if key:
+            if key in self.cache:
+                del self.cache[key]
+                del self.current_size[key]
+        else:
+            self.cache.clear()
+            self.current_size.clear()

aws_lambda_powertools/logging/buffer/config.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+from typing import Literal
+
+
+class LoggerBufferConfig:
+    """
+    Configuration for log buffering behavior.
+    """
+
+    # Define a type alias for valid log levels
+    LogLevelType = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
+
+    # Define class-level constant for valid log levels
+    VALID_LOG_LEVELS: list[str] = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
+
+    def __init__(
+        self,
+        max_size: int = 10240,
+        minimum_log_level: LogLevelType = "DEBUG",
+        flush_on_error: bool = True,
+        compress: bool = False,
+    ):
+        """
+        Initialize logger buffer configuration.
+
+        Parameters
+        ----------
+        max_size : int, optional
+            Maximum size of the buffer in bytes
+        minimum_log_level : str, optional
+            Minimum log level to buffer
+        flush_on_error : bool, optional
+            Whether to flush the buffer when an error occurs
+        compress : bool, optional
+            Whether to compress buffered logs
+        """
+        self._validate_inputs(max_size, minimum_log_level, flush_on_error, compress)
+
+        self._max_size = max_size
+        self._minimum_log_level = minimum_log_level.upper()
+        self._flush_on_error = flush_on_error
+        self._compress = compress
+
+    def _validate_inputs(
+        self,
+        max_size: int,
+        minimum_log_level: str,
+        flush_on_error: bool,
+        compress: bool,
+    ) -> None:
+        """
+        Validate configuration inputs.
+
+        Parameters
+        ----------
+        Same as __init__ method parameters
+        """
+        if not isinstance(max_size, int) or max_size <= 0:
+            raise ValueError("Max size must be a positive integer")
+
+        if not isinstance(minimum_log_level, str):
+            raise ValueError("Log level must be a string")
+
+        # Validate log level
+        if minimum_log_level.upper() not in self.VALID_LOG_LEVELS:
+            raise ValueError(f"Invalid log level. Must be one of {self.VALID_LOG_LEVELS}")
+
+        if not isinstance(flush_on_error, bool):
+            raise ValueError("flush_on_error must be a boolean")
+
+        if not isinstance(compress, bool):
+            raise ValueError("compress must be a boolean")
+
+    @property
+    def max_size(self) -> int:
+        """Maximum buffer size in bytes."""
+        return self._max_size
+
+    @property
+    def minimum_log_level(self) -> str:
+        """Minimum log level to buffer."""
+        return self._minimum_log_level
+
+    @property
+    def flush_on_error(self) -> bool:
+        """Flag to flush buffer on error."""
+        return self._flush_on_error
+
+    @property
+    def compress(self) -> bool:
+        """Flag to compress buffered logs."""
+        return self._compress

tests/unit/logger/__init__.py

@@ -0,0 +1,166 @@
+import pytest
+
+from aws_lambda_powertools.logging.buffer import LoggerBufferCache
+from aws_lambda_powertools.warnings import PowertoolsUserWarning
+
+
+def test_initialization():
+
+    # GIVEN a new instance of LoggerBufferCache
+    logger_cache = LoggerBufferCache(1000)
+
+    # THEN cache should have correct initial state
+    assert logger_cache.max_size_bytes == 1000
+    assert logger_cache.cache == {}
+    assert logger_cache.current_size == {}
+
+
+def test_add_single_item():
+    # GIVEN a new instance of LoggerBufferCache with 1024 bytes
+    logger_cache = LoggerBufferCache(1024)
+
+    # WHEN a single item is added
+    logger_cache.add("key1", "test_item")
+
+    # THEN item is stored correctly with proper size tracking
+    assert len(logger_cache.get("key1")) == 1
+    assert logger_cache.get("key1")[0] == "test_item"
+    assert logger_cache.current_size["key1"] == len("test_item")
+
+
+def test_add_multiple_items_same_key():
+    # GIVEN a new instance of LoggerBufferCache with 1024 bytes
+    logger_cache = LoggerBufferCache(1024)
+
+    # WHEN multiple items are added to the same key
+    logger_cache.add("key1", "item1")
+    logger_cache.add("key1", "item2")
+
+    # THEN items are stored sequentially
+    assert len(logger_cache.get("key1")) == 2
+    assert logger_cache.get("key1") == ["item1", "item2"]
+
+
+def test_cache_size_limit_single_key():
+    # GIVEN a new instance of LoggerBufferCache with small cache size
+    logger_cache = LoggerBufferCache(10)
+
+    # WHEN multiple items are added
+    logger_cache.add("key1", "long_item1")
+    logger_cache.add("key1", "long_item2")
+    logger_cache.add("key1", "long_item3")
+
+    # THEN cache maintains size limit for a single key
+    assert len(logger_cache.get("key1")) > 0
+    assert logger_cache.current_size["key1"] <= 10
+
+
+def test_item_larger_than_cache():
+    # GIVEN a new instance of LoggerBufferCache with small cache size
+    logger_cache = LoggerBufferCache(5)
+
+    # WHEN an item larger than cache is added
+    with pytest.warns(expected_warning=PowertoolsUserWarning, match="Item size *"):
+        # THEN a warning is raised
+        logger_cache.add("key1", "very_long_item")
+
+    # THEN the key is not added
+    assert "key1" not in logger_cache.cache
+
+
+def test_get_existing_key():
+    # GIVEN a new instance of LoggerBufferCache with 1024 bytes
+    logger_cache = LoggerBufferCache(1024)
+
+    # WHEN we add keys
+    logger_cache.add("key1", "item1")
+    logger_cache.add("key1", "item2")
+
+    # THEN all items are retrieved
+    assert logger_cache.get("key1") == ["item1", "item2"]
+
+
+def test_get_non_existing_key():
+    # GIVEN a new instance of LoggerBufferCache with 1024 bytes
+    logger_cache = LoggerBufferCache(1000)
+
+    # WHEN getting items for a non-existing key
+    retrieved = logger_cache.get("non_existing")
+
+    # THEN an empty list is returned
+    assert retrieved == []
+
+
+def test_clear_all():
+    # GIVEN a new instance of LoggerBufferCache with 1024 bytes
+    logger_cache = LoggerBufferCache(1024)
+
+    # WHEN we add multiple keys
+    logger_cache.add("key1", "item1")
+    logger_cache.add("key2", "item2")
+
+    # WHEN clearing all keys
+    logger_cache.clear()
+
+    # THEN cache becomes empty
+    assert logger_cache.cache == {}
+    assert logger_cache.current_size == {}
+
+
+def test_clear_specific_key():
+    # GIVEN a new instance of LoggerBufferCache with 1024 bytes
+    logger_cache = LoggerBufferCache(1024)
+
+    # WHEN we add multiple keys
+    logger_cache.add("key1", "item1")
+    logger_cache.add("key2", "item2")
+
+    # WHEN we remove a specific key
+    logger_cache.clear("key1")
+
+    # THEN only that key is removed
+    assert "key1" not in logger_cache.cache
+    assert "key2" in logger_cache.cache
+    assert logger_cache.get("key1") == []
+
+
+def test_multiple_keys_with_size_limits():
+    # GIVEN a new instance of LoggerBufferCache with 20 bytes
+    logger_cache = LoggerBufferCache(20)
+
+    # WHEN adding items to multiple keys
+    logger_cache.add("key1", "item1")
+    logger_cache.add("key1", "item2")
+    logger_cache.add("key2", "long_item")
+
+    # THEN total size remains within limit
+    assert len(logger_cache.cache["key1"]) > 0
+    assert len(logger_cache.cache["key2"]) > 0
+    assert logger_cache.current_size["key1"] + logger_cache.current_size["key2"] <= 20
+
+
+def test_add_different_types():
+    # GIVEN a new instance of LoggerBufferCache with 1024 bytes
+    logger_cache = LoggerBufferCache(1024)
+
+    # WHEN adding items of different types
+    logger_cache.add("key1", 123)
+    logger_cache.add("key1", [1, 2, 3])
+    logger_cache.add("key1", {"a": 1})
+
+    # THEN items are stored successfully
+    retrieved = logger_cache.get("key1")
+    assert len(retrieved) == 3
+
+
+def test_cache_size_tracking():
+    # GIVEN a new instance of LoggerBufferCache with 30 bytes
+    logger_cache = LoggerBufferCache(30)
+
+    # WHEN adding items
+    logger_cache.add("key1", "small")
+    logger_cache.add("key1", "another_item")
+
+    # THEN current size is tracked correctly
+    assert logger_cache.current_size["key1"] == len("small") + len("another_item")
+    assert logger_cache.current_size["key1"] <= 30