Logging πŸ““οƒ

Importing and creating a logging instance πŸ§ͺ

[ ]:

from __future__ import annotations

import structlog
from peak.tools.logging import LogHandler, LogLevel, PeakLogger, get_logger

Example 1: Get a default logger

[ ]:

logger: PeakLogger = get_logger(level=LogLevel.DEBUG)

logger.info("structured logging is easy", config=structlog.get_config())
# gives you: {"config":{"cache_logger_on_first_use":true,"context_class":"<class 'dict'>","logger_factory":"<structlog.stdlib.LoggerFactory object at 0x1035fbbe0>","processors":["<function merge_contextvars at 0x103001ab0>","<function pii_masking_processor at 0x1078db9a0>","<function peak_contexts_processor at 0x107823250>","<function filter_by_level at 0x103414a60>","<function add_logger_name at 0x1034155a0>","<function add_log_level at 0x103002560>","<structlog.stdlib.PositionalArgumentsFormatter object at 0x1035f91e0>","<structlog.processors.TimeStamper object at 0x1078f25c0>","<structlog.processors.StackInfoRenderer object at 0x1035f9150>","<structlog.processors.ExceptionRenderer object at 0x102322da0>","<structlog.processors.UnicodeDecoder object at 0x1035f9a20>","<structlog.processors.EventRenamer object at 0x1035fb400>","<structlog.processors.JSONRenderer object at 0x1035f9270>"],"wrapper_class":"<class 'structlog._native.BoundLoggerFilteringAtDebug'>"},"level":"info","logger":"root","message":"structured logging is easy","source":"peak-sdk","timestamp":"2024-03-15T18:32:24.758860Z"}

logger.debug("This is an info message")
# gives you: {"level":"debug","logger":"root","message":"This is an info message","source":"peak-sdk","timestamp":"2024-03-15T18:32:24.762185Z"}

Example 2: Get a customized logger

[ ]:

logger: PeakLogger = get_logger(
    name=__name__,
    level=LogLevel.ERROR,
    handlers=[LogHandler.CONSOLE],
)

logger.error("This is an error message")
# gives you: {"level":"error","logger":"__main__","message":"This is an error message","source":"peak-sdk","timestamp":"2024-03-15T18:33:28.749419Z"}

logger.info("This is an info message and hence will not be printed")
# gives you: nothing

Masking sensitive data πŸ§ͺ

Example 1: Masking is enabled by default

[ ]:

logger: PeakLogger = get_logger(__name__)

logger.error("This is my email user@someorg.com")
# gives you: {"level":"error","logger":"__main__","message":"This is my email use****************com","source":"peak-sdk","timestamp":"2024-03-15T18:33:01.000308Z"}

Example 2: Disabling masking

[ ]:

logger: PeakLogger = get_logger(name=__name__, disable_masking=True)

logger.error("This is my email user@someorg.com")
# gives you: {"level":"error","logger":"__main__","message":"This is my email user@someorg.com","source":"peak-sdk","timestamp":"2024-04-10T06:25:47.703135Z"}

Binding and unbinding contexts πŸ§ͺ

Example 1: Bind additional context to logger

[ ]:

logger: PeakLogger = get_logger(
    name=__name__,
)

# bind the context
logger.bind({"key1": "value1", "key2": "value2"})

logger.info("Info msg with additional context")
# gives you: {"key1":"value1","key2":"value2","level":"info","logger":"__main__","message":"Info msg with additional context","source":"peak-sdk","timestamp":"2024-04-10T06:26:08.271239Z"}

Example 2: Unbind key(s) from the context

[ ]:

logger: PeakLogger = get_logger(
    name=__name__,
)

# bind the context
logger.bind({"key1": "value1", "key2": "value2"})

logger.info("Info msg with additional context")
# gives you: {"key1":"value1","key2":"value2","level":"info","logger":"__main__","message":"Info msg with additional context","source":"peak-sdk","timestamp":"2024-04-10T06:26:24.113639Z"}

# unbind key1 from the context
logger.unbind(["key1", "key3"])

logger.info("Info msg with partial context")
# gives you: {"key2":"value2","level":"info","logger":"__main__","message":"Info msg with partial context","source":"peak-sdk","timestamp":"2024-04-10T06:26:24.115962Z"}

Changing log level πŸ§ͺ

Example 1: Change or Set the log level

[ ]:

logger: PeakLogger = get_logger(
    name=__name__,
    level=LogLevel.INFO,
)

# log level is INFO
logger.info("This is the first info msg")
# gives you: {"level":"info","logger":"__main__","message":"This is the first info msg","source":"peak-sdk","timestamp":"2024-03-15T18:34:33.646110Z"}

logger.error("This is the first error msg")
# gives you: {"level":"error","logger":"__main__","message":"This is the first error msg","source":"peak-sdk","timestamp":"2024-03-15T18:34:33.648278Z"}

# change the log level to ERROR
logger.set_log_level(LogLevel.ERROR)

logger.info("This is the second info msg hence will not be printed")
# gives you : nothing

logger.error("This is the second error msg")
# gives you : {"level":"error","logger":"__main__","message":"This is the second error msg","source":"peak-sdk","timestamp":"2024-03-15T18:34:33.651846Z"}

Additional examples πŸ§ͺ

Example 1: Using Formatting Placeholders

[ ]:

logger: PeakLogger = get_logger()

name = "John"
age = 30

logger.info("User: %s, Age: %d", name, age)
# gives you: {"level":"info","logger":"root","message":"User: John, Age: 30","source":"peak-sdk","timestamp":"2024-04-10T06:27:03.176011Z"}

Example 2: Logging with Structured Data

[ ]:

from typing import Any

logger: PeakLogger = get_logger()

data: Any = {"user_id": 123, "action": "Login"}

logger.info("User performed an action", data=data)
# gives you: {"data":{"action":"Login","user_id":123},"level":"info","logger":"root","message":"User performed an action","source":"peak-sdk","timestamp":"2024-04-10T06:27:16.466597Z"}

Example 3: Logging Mixed Data Types

[ ]:

logger: PeakLogger = get_logger()

user_id = 123
action = "Login"
result: Any = {"success": True, "message": "Logged in successfully"}

logger.info("Logging mixed data types", user_id=user_id, action=action, result=result)
# gives you: {"action":"Login","level":"info","logger":"root","message":"Logging mixed data types","result":{"message":"Logged in successfully","success":true},"source":"peak-sdk","timestamp":"2024-04-10T06:27:24.513378Z","user_id":123}

Example 4: Logging Exceptions

[ ]:

logger: PeakLogger = get_logger()

try:
    result: float = 10 / 0
except Exception:
    logger.exception("An exception occurred")

# gives you:
# {"exception":"Traceback (most recent call last):\n  File \"/var/folders/61/qg8m95nx2bl4zqyccrc27bvw0000gp/T/ipykernel_84610/948709948.py\", line 4, in <module>\n    result: float = 10 / 0\nZeroDivisionError: division by zero","level":"error","logger":"root","message":"An exception occurred","source":"peak-sdk","timestamp":"2024-04-10T06:27:36.005146Z"}
# Traceback (most recent call last):
#   File "/var/folders/**/ipykernel_84610/**.py", line 4, in <module>
#     result: float = 10 / 0
# ZeroDivisionError: division by zero

Example 5: Logging with custom / user-providable processors

[ ]:

from typing import List


def simple_no_tty_custom_processors_factory() -> List[structlog.types.Processor | Any]:
    """Simple custom processors factory function that returns a list of processors.

    Knowingly, this function does not include the EventRenamer processor to demonstrate how it will be patched in if missing.

    Returns:
        List[structlog.types.Processor | Any]: List of processors
    """
    return [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ]


custom_logger: PeakLogger = get_logger(
    custom_processors_factory=simple_no_tty_custom_processors_factory,
)

logger.info(
    "This custom logger was patched to include EventRenamer processors",
    processors_list=structlog.get_config()["processors"],
)
# gives you: {"level":"info","logger":"__main__","message":"This custom logger was patched to include EventRenamer processors","processors_list":["<function merge_contextvars at 0x103001ab0>","<function filter_by_level at 0x103414a60>","<function add_logger_name at 0x1034155a0>","<function add_log_level at 0x103002560>","<structlog.stdlib.PositionalArgumentsFormatter object at 0x1035fbeb0>","<structlog.processors.TimeStamper object at 0x107913940>","<structlog.processors.EventRenamer object at 0x1035fb490>","<structlog.processors.JSONRenderer object at 0x1035f9210>"],"source":"peak-sdk","timestamp":"2024-03-15T18:35:36.971465Z"}

logger.info("This is a custom logger")
# gives you: {"level":"info","logger":"__main__","message":"This is a custom logger","source":"peak-sdk","timestamp":"2024-03-15T18:35:36.973325Z"}

Clone an existing logger

[ ]:

logger: PeakLogger = get_logger("my_clone-able_logger")

logger.set_log_level(LogLevel.DEBUG)
logger.info("This is a clone-able logger")
# gives you: {"level":"info","logger":"my_clone-able_logger","message":"This is a clone-able logger","source":"peak-sdk","timestamp":"2024-03-18T11:49:35.696540Z"}

logger.bind({"key1": "value1", "key2": "value2"})

cloned_logger: PeakLogger = logger.clone_with_context({"zoo": "keeper"})

logger.debug("This is the original logger")
# gives you: {"level":"debug","logger":"my_clone-able_logger","message":"This is the original logger","source":"peak-sdk","timestamp":"2024-03-18T11:49:35.699316Z"}

cloned_logger.debug(
    "This is a cloned logger that has all the context and log level as the original logger along with the additional context we added",
)
# gives you: {"level":"debug","logger":"my_clone-able_logger","message":"This is a cloned logger that has all the context and log level as the original logger along with the additional context we added","source":"peak-sdk","timestamp":"2024-03-18T11:54:22.816122Z","zoo":"keeper"}