Skip to content

log

The log engine.

CatLogEngine

The log engine.

Engine to filter the logs in the terminal according to the level of severity.

Attributes:

Name Type Description
LOG_LEVEL str

Level of logging set in the .env file.
Notes

The logging level set in the .env file will print all the logs from that level to above. Available levels are:

- `DEBUG`
- `INFO`
- `WARNING`
- `ERROR`
- `CRITICAL`

Default to INFO.

Source code in cat/log.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 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
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
class CatLogEngine:
    """The log engine.

    Engine to filter the logs in the terminal according to the level of severity.

    Attributes
    ----------
    LOG_LEVEL : str
        Level of logging set in the `.env` file.

    Notes
    -----
    The logging level set in the `.env` file will print all the logs from that level to above.
    Available levels are:

        - `DEBUG`
        - `INFO`
        - `WARNING`
        - `ERROR`
        - `CRITICAL`

    Default to `INFO`.

    """

    def __init__(self):
        self.LOG_LEVEL = get_log_level()
        self.default_log()

        # workaround for pdfminer logging
        # https://github.com/pdfminer/pdfminer.six/issues/347
        logging.getLogger("pdfminer").setLevel(logging.WARNING)

    def show_log_level(self, record):
        """Allows to show stuff in the log based on the global setting.

        Parameters
        ----------
        record : dict

        Returns
        -------
        bool

        """
        return record["level"].no >= logger.level(self.LOG_LEVEL).no

    def default_log(self):
        """Set the same debug level to all the project dependencies.

        Returns
        -------
        """

        time = "<green>[{time:YYYY-MM-DD HH:mm:ss.SSS}]</green>"
        level = "<level>{level: <6}</level>"
        origin = "<level>{extra[original_name]}.{extra[original_class]}.{extra[original_caller]}::{extra[original_line]}</level>"
        message = "<level>{message}</level>"
        log_format = f"{time} {level} {origin} \n{message}"

        logger.remove()
        if self.LOG_LEVEL == "DEBUG":
            return logger.add(
                sys.stdout,
                colorize=True,
                format=log_format,
                backtrace=True,
                diagnose=True,
                filter=self.show_log_level,
            )
        else:
            return logger.add(
                sys.stdout,
                colorize=True,
                format=log_format,
                filter=self.show_log_level,
                level=self.LOG_LEVEL,
            )

    def get_caller_info(self, skip=3):
        """Get the name of a caller in the format module.class.method.

        Copied from: https://gist.github.com/techtonik/2151727

        Parameters
        ----------
        skip :  int
            Specifies how many levels of stack to skip while getting caller name.

        Returns
        -------
        package : str
            Caller package.
        module : str
            Caller module.
        klass : str
            Caller classname if one otherwise None.
        caller : str
            Caller function or method (if a class exist).
        line : int
            The line of the call.


        Notes
        -----
        skip=1 means "who calls me",
        skip=2 "who calls my caller" etc.

        An empty string is returned if skipped levels exceed stack height.
        """
        stack = inspect.stack()
        start = 0 + skip
        if len(stack) < start + 1:
            return ""
        parentframe = stack[start][0]

        # module and packagename.
        module_info = inspect.getmodule(parentframe)
        if module_info:
            mod = module_info.__name__.split(".")
            package = mod[0]
            module = ".".join(mod[1:])

        # class name.
        klass = ""
        if "self" in parentframe.f_locals:
            klass = parentframe.f_locals["self"].__class__.__name__

        # method or function name.
        caller = None
        if parentframe.f_code.co_name != "<module>":  # top level usually
            caller = parentframe.f_code.co_name

        # call line.
        line = parentframe.f_lineno

        # Remove reference to frame
        # See: https://docs.python.org/3/library/inspect.html#the-interpreter-stack
        del parentframe

        return package, module, klass, caller, line

    def __call__(self, msg, level="DEBUG"):
        """Alias of self.log()"""
        self.log(msg, level)

    def debug(self, msg):
        """Logs a DEBUG message"""
        self.log(msg, level="DEBUG")

    def info(self, msg):
        """Logs an INFO message"""
        self.log(msg, level="INFO")

    def warning(self, msg):
        """Logs a WARNING message"""
        self.log(msg, level="WARNING")

    def error(self, msg):
        """Logs an ERROR message"""
        self.log(msg, level="ERROR")

    def critical(self, msg):
        """Logs a CRITICAL message"""
        self.log(msg, level="CRITICAL")

    def log(self, msg, level="DEBUG"):
        """Log a message

        Parameters
        ----------
        msg :
            Message to be logged.
        level : str
            Logging level."""

        (package, module, klass, caller, line) = self.get_caller_info()

        custom_logger = logger.bind(
            original_name=f"{package}.{module}",
            original_line=line,
            original_class=klass,
            original_caller=caller,
        )

        # prettify
        if type(msg) in [dict, list, str]:  # TODO: should be recursive
            try:
                msg = json.dumps(msg, indent=4)
            except Exception:
                pass
        else:
            msg = pformat(msg)

        # actual log
        custom_logger.log(level, msg)

    def welcome(self):
        """Welcome message in the terminal."""
        secure = "s" if get_env("CCAT_CORE_USE_SECURE_PROTOCOLS") in ("true", "1") else ""

        cat_host = get_env("CCAT_CORE_HOST")
        cat_port = get_env("CCAT_CORE_PORT")
        cat_address = f"http{secure}://{cat_host}:{cat_port}"

        with open("cat/welcome.txt", "r") as f:
            print(f.read())

        print("\n=============== ^._.^ ===============\n")
        print(f"Cat REST API:   {cat_address}/docs")
        print(f"Cat ADMIN:      {cat_address}/admin\n")
        print("======================================")

__call__(msg, level='DEBUG')

Alias of self.log()

Source code in cat/log.py 
160
161
162
def __call__(self, msg, level="DEBUG"):
    """Alias of self.log()"""
    self.log(msg, level)

critical(msg)

Logs a CRITICAL message

Source code in cat/log.py 
180
181
182
def critical(self, msg):
    """Logs a CRITICAL message"""
    self.log(msg, level="CRITICAL")

debug(msg)

Logs a DEBUG message

Source code in cat/log.py 
164
165
166
def debug(self, msg):
    """Logs a DEBUG message"""
    self.log(msg, level="DEBUG")

default_log()

Set the same debug level to all the project dependencies.

Source code in cat/log.py 
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
91
92
93
94
95
def default_log(self):
    """Set the same debug level to all the project dependencies.

    Returns
    -------
    """

    time = "<green>[{time:YYYY-MM-DD HH:mm:ss.SSS}]</green>"
    level = "<level>{level: <6}</level>"
    origin = "<level>{extra[original_name]}.{extra[original_class]}.{extra[original_caller]}::{extra[original_line]}</level>"
    message = "<level>{message}</level>"
    log_format = f"{time} {level} {origin} \n{message}"

    logger.remove()
    if self.LOG_LEVEL == "DEBUG":
        return logger.add(
            sys.stdout,
            colorize=True,
            format=log_format,
            backtrace=True,
            diagnose=True,
            filter=self.show_log_level,
        )
    else:
        return logger.add(
            sys.stdout,
            colorize=True,
            format=log_format,
            filter=self.show_log_level,
            level=self.LOG_LEVEL,
        )

error(msg)

Logs an ERROR message

Source code in cat/log.py 
176
177
178
def error(self, msg):
    """Logs an ERROR message"""
    self.log(msg, level="ERROR")

get_caller_info(skip=3)

Get the name of a caller in the format module.class.method.

Copied from: https://gist.github.com/techtonik/2151727

Parameters:

Name Type Description Default
skip int

Specifies how many levels of stack to skip while getting caller name.

 3

Returns:

Name Type Description
package str

Caller package.
module str

Caller module.
klass str

Caller classname if one otherwise None.
caller str

Caller function or method (if a class exist).
line int

The line of the call.
Notes

skip=1 means "who calls me", skip=2 "who calls my caller" etc.

An empty string is returned if skipped levels exceed stack height.

Source code in cat/log.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def get_caller_info(self, skip=3):
    """Get the name of a caller in the format module.class.method.

    Copied from: https://gist.github.com/techtonik/2151727

    Parameters
    ----------
    skip :  int
        Specifies how many levels of stack to skip while getting caller name.

    Returns
    -------
    package : str
        Caller package.
    module : str
        Caller module.
    klass : str
        Caller classname if one otherwise None.
    caller : str
        Caller function or method (if a class exist).
    line : int
        The line of the call.


    Notes
    -----
    skip=1 means "who calls me",
    skip=2 "who calls my caller" etc.

    An empty string is returned if skipped levels exceed stack height.
    """
    stack = inspect.stack()
    start = 0 + skip
    if len(stack) < start + 1:
        return ""
    parentframe = stack[start][0]

    # module and packagename.
    module_info = inspect.getmodule(parentframe)
    if module_info:
        mod = module_info.__name__.split(".")
        package = mod[0]
        module = ".".join(mod[1:])

    # class name.
    klass = ""
    if "self" in parentframe.f_locals:
        klass = parentframe.f_locals["self"].__class__.__name__

    # method or function name.
    caller = None
    if parentframe.f_code.co_name != "<module>":  # top level usually
        caller = parentframe.f_code.co_name

    # call line.
    line = parentframe.f_lineno

    # Remove reference to frame
    # See: https://docs.python.org/3/library/inspect.html#the-interpreter-stack
    del parentframe

    return package, module, klass, caller, line

info(msg)

Logs an INFO message

Source code in cat/log.py 
168
169
170
def info(self, msg):
    """Logs an INFO message"""
    self.log(msg, level="INFO")

log(msg, level='DEBUG')

Log a message

Parameters:

Name Type Description Default
msg

Message to be logged.

 required
level str

Logging level.

 'DEBUG'
Source code in cat/log.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def log(self, msg, level="DEBUG"):
    """Log a message

    Parameters
    ----------
    msg :
        Message to be logged.
    level : str
        Logging level."""

    (package, module, klass, caller, line) = self.get_caller_info()

    custom_logger = logger.bind(
        original_name=f"{package}.{module}",
        original_line=line,
        original_class=klass,
        original_caller=caller,
    )

    # prettify
    if type(msg) in [dict, list, str]:  # TODO: should be recursive
        try:
            msg = json.dumps(msg, indent=4)
        except Exception:
            pass
    else:
        msg = pformat(msg)

    # actual log
    custom_logger.log(level, msg)

show_log_level(record)

Allows to show stuff in the log based on the global setting.

Parameters:

Name Type Description Default
record dict
required

Returns:

Type Description
bool
Source code in cat/log.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
def show_log_level(self, record):
    """Allows to show stuff in the log based on the global setting.

    Parameters
    ----------
    record : dict

    Returns
    -------
    bool

    """
    return record["level"].no >= logger.level(self.LOG_LEVEL).no

warning(msg)

Logs a WARNING message

Source code in cat/log.py 
172
173
174
def warning(self, msg):
    """Logs a WARNING message"""
    self.log(msg, level="WARNING")

welcome()

Welcome message in the terminal.

Source code in cat/log.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def welcome(self):
    """Welcome message in the terminal."""
    secure = "s" if get_env("CCAT_CORE_USE_SECURE_PROTOCOLS") in ("true", "1") else ""

    cat_host = get_env("CCAT_CORE_HOST")
    cat_port = get_env("CCAT_CORE_PORT")
    cat_address = f"http{secure}://{cat_host}:{cat_port}"

    with open("cat/welcome.txt", "r") as f:
        print(f.read())

    print("\n=============== ^._.^ ===============\n")
    print(f"Cat REST API:   {cat_address}/docs")
    print(f"Cat ADMIN:      {cat_address}/admin\n")
    print("======================================")

get_log_level()

Return the global LOG level.

Source code in cat/log.py 
13
14
15
def get_log_level():
    """Return the global LOG level."""
    return get_env("CCAT_LOG_LEVEL")