Skip to content

logger

This module defines the logging interface for storing system logs.

__all__ = ['logger', 'init', 'default_log_filter'] module-attribute

default_log_filter = None module-attribute

module_names = [name for (_, name, _) in pkgutil.iter_modules(['roc'])] module-attribute

DebugModuleLevel

Bases: BaseModel

A Pydantic model for storing the logging level. Used primarily for setting per-module logging levels.

Source code in roc/logger.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class DebugModuleLevel(BaseModel):
    """A Pydantic model for storing the logging level. Used primarily for
    setting per-module logging levels.
    """

    module_name: str
    log_level: str = Field(pattern=r"TRACE|DEBUG|INFO|SUCCESS|WARNING|ERROR|CRITICAL")

    @field_validator("module_name", mode="before")
    @classmethod
    def validate_module_name(cls, name: str) -> str:
        assert (
            name in module_names
        ), f"Module name '{name}' not a valid module name. Must be one of {module_names}"

        return name

log_level = Field(pattern='TRACE|DEBUG|INFO|SUCCESS|WARNING|ERROR|CRITICAL') class-attribute instance-attribute

module_name instance-attribute

validate_module_name(name) classmethod

Source code in roc/logger.py
25
26
27
28
29
30
31
32
@field_validator("module_name", mode="before")
@classmethod
def validate_module_name(cls, name: str) -> str:
    assert (
        name in module_names
    ), f"Module name '{name}' not a valid module name. Must be one of {module_names}"

    return name

LogFilter

A Callable filter that determines if the loguru record should be logged or not.

Source code in roc/logger.py
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class LogFilter:
    """A Callable filter that determines if the loguru record should be logged
    or not.
    """

    def __init__(
        self,
        *,
        level: str | None = None,
        log_modules: str | None = None,
        enabled: bool = True,
        use_module_settings: bool = True,
    ):
        settings = Config.get()
        self.level = level or settings.log_level
        self.level_num = logger.level(self.level).no
        if not isinstance(log_modules, str):
            if use_module_settings:
                log_modules = settings.log_modules
            else:
                log_modules = ""
        mod_list = self.parse_module_str(log_modules)
        self.module_levels = {mod_lvl.module_name: mod_lvl.log_level for mod_lvl in mod_list}

    def __call__(self, record: Any) -> bool:
        # TODO: this would be more effecient as a dict rather than a loop (O(1) rather than O(n))

        if record["module"] in self.module_levels:
            mod_log_level = self.module_levels[record["module"]]
            mod_level_num = logger.level(mod_log_level).no
            if record["level"].no >= mod_level_num:
                return True
            else:
                return False

        if record["level"].no >= self.level_num:
            return True

        return False

    @classmethod
    def parse_module_str(cls, s: str) -> list[DebugModuleLevel]:
        s = s.strip()

        mod_list = s.split(";")
        # empty str
        if mod_list == [""]:
            return []

        mod_lvl_list: list[dict[str, str]] = []
        for mod in mod_list:
            mod_parts = mod.split(":")
            mod_lvl_list.append({"module_name": mod_parts[0], "log_level": mod_parts[1]})

        debug_module_list = TypeAdapter(list[DebugModuleLevel])
        return debug_module_list.validate_python(mod_lvl_list)

level = level or settings.log_level instance-attribute

level_num = logger.level(self.level).no instance-attribute

module_levels = {mod_lvl.module_name: mod_lvl.log_levelfor mod_lvl in mod_list} instance-attribute

__call__(record)

Source code in roc/logger.py
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __call__(self, record: Any) -> bool:
    # TODO: this would be more effecient as a dict rather than a loop (O(1) rather than O(n))

    if record["module"] in self.module_levels:
        mod_log_level = self.module_levels[record["module"]]
        mod_level_num = logger.level(mod_log_level).no
        if record["level"].no >= mod_level_num:
            return True
        else:
            return False

    if record["level"].no >= self.level_num:
        return True

    return False

__init__(*, level=None, log_modules=None, enabled=True, use_module_settings=True)

Source code in roc/logger.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    *,
    level: str | None = None,
    log_modules: str | None = None,
    enabled: bool = True,
    use_module_settings: bool = True,
):
    settings = Config.get()
    self.level = level or settings.log_level
    self.level_num = logger.level(self.level).no
    if not isinstance(log_modules, str):
        if use_module_settings:
            log_modules = settings.log_modules
        else:
            log_modules = ""
    mod_list = self.parse_module_str(log_modules)
    self.module_levels = {mod_lvl.module_name: mod_lvl.log_level for mod_lvl in mod_list}

parse_module_str(s) classmethod

Source code in roc/logger.py
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@classmethod
def parse_module_str(cls, s: str) -> list[DebugModuleLevel]:
    s = s.strip()

    mod_list = s.split(";")
    # empty str
    if mod_list == [""]:
        return []

    mod_lvl_list: list[dict[str, str]] = []
    for mod in mod_list:
        mod_parts = mod.split(":")
        mod_lvl_list.append({"module_name": mod_parts[0], "log_level": mod_parts[1]})

    debug_module_list = TypeAdapter(list[DebugModuleLevel])
    return debug_module_list.validate_python(mod_lvl_list)

init()

Initializes the logging module. Installs the filter, fetches the settings, etc.

Source code in roc/logger.py
 96
 97
 98
 99
100
101
102
103
104
105
106
def init() -> None:
    """Initializes the logging module. Installs the filter, fetches the
    settings, etc.
    """
    global default_log_filter
    default_log_filter = LogFilter()

    logger.remove()
    settings = Config.get()
    if settings.log_enable:
        logger.add(sys.stderr, level=0, filter=default_log_filter)