Logging πŸ““οƒ

Importing and creating a logging instance πŸ§ͺ

[ ]:

from __future__ import annotations  # Needed mainly for advanced logging

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

Example 1: Get a default logger

[ ]:

logger: PeakLogger = get_logger()

logger.info("This is an info message")
# gives you: {"level":"info","logger":"root","message":"This is an info message","source":"peak-sdk","timestamp":"2024-08-07T09:25:30.810668Z"}

logger.debug("This is a debug message hence will not be printed")
# Gives you: nothing! Since the default LogLevel for the logger is INFO, only messages at the INFO level and higher (e.g., ERROR) are printed.

logger.error("This is an error message")
# Gives you: {"level":"error","logger":"root","message":"This is an error message","source":"peak-sdk","timestamp":"2024-08-07T09:25:30.812268Z"}

Example 2: Get a customized logger

[ ]:

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

logger.debug("This is an debug message")
# gives you: {"level":"debug","logger":"__main__","message":"This is an debug message","source":"peak-sdk","timestamp":"2024-08-07T09:26:15.694022Z"}

logger.info("This is an info message")
# gives you: {"level":"info","logger":"__main__","message":"This is an info message","source":"peak-sdk","timestamp":"2024-08-07T09:26:15.694721Z"}

logger.error("This is an error message")
# gives you: {"level":"error","logger":"__main__","message":"This is an error message","source":"peak-sdk","timestamp":"2024-08-07T09:26:15.695818Z"}

Masking sensitive data πŸ§ͺ

Example 1: Masking is enabled by default

[ ]:

logger: PeakLogger = get_logger(__name__)

logger.info("This is my email user@someorg.com")
# gives you: {"level":"info","logger":"__main__","message":"This is my email use****************com","source":"peak-sdk","timestamp":"2024-08-07T09:27:34.972619Z"}

Example 2: Disabling masking

[ ]:

logger: PeakLogger = get_logger(
    name=__name__,
    disable_masking=True,  # not recommended for production
)

logger.info("This is my email user@someorg.com")
# gives you: {"level":"info","logger":"__main__","message":"This is my email user@someorg.com","source":"peak-sdk","timestamp":"2024-08-07T09:28:00.310961Z"}

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-08-07T09:28:25.615493Z"}

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-08-07T09:28:46.414069Z"}

# 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-08-07T09:28:46.415456Z"}

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-08-07T09:30:05.547999Z"}

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-08-07T09:30:05.550725Z"}

# 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! since LogLevel is now updated to ERROR

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-08-07T09:30:05.555716Z"}

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

import structlog


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"}

Example 6: 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"}