Skip to content

glide_sync.logger

Logger

A singleton class that allows logging which is consistent with logs from the internal GLIDE core. The logger can be set up in 2 ways - 1. By calling Logger.init, which configures the logger only if it wasn't previously configured. 2. By calling Logger.set_logger_config, which replaces the existing configuration, and means that new logs will not be saved with the logs that were sent before the call. If none of these functions are called, the first log attempt will initialize a new logger with default configuration.

Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/logger.py 
 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
class Logger:
    """
    A singleton class that allows logging which is consistent with logs from the internal GLIDE core.
    The logger can be set up in 2 ways -
    1. By calling Logger.init, which configures the logger only if it wasn't previously configured.
    2. By calling Logger.set_logger_config, which replaces the existing configuration, and means that new logs will not be
    saved with the logs that were sent before the call.
    If none of these functions are called, the first log attempt will initialize a new logger with default configuration.
    """

    _instance: Logger | None = None
    _glide_ffi = _GlideFFI()
    _ffi = _glide_ffi.ffi
    _lib = _glide_ffi.lib
    logger_level: Level = Level.OFF

    def __init__(self, level: Optional[Level] = None, file_name: Optional[str] = None):
        c_level = (
            Logger._ffi.new("Level*", level.value)
            if level is not None
            else Logger._ffi.NULL
        )
        c_file_name = (
            Logger._ffi.new("char[]", file_name.encode(ENCODING))
            if file_name
            else Logger._ffi.NULL
        )

        result_ptr = Logger._lib.init(c_level, c_file_name)

        if result_ptr != Logger._ffi.NULL:
            try:
                if result_ptr.log_error != Logger._ffi.NULL:
                    error_str = cast(
                        bytes, Logger._ffi.string(result_ptr.log_error)
                    ).decode(ENCODING)
                    raise LoggerError(f"Logger initialization failed: {error_str}")
                else:
                    Logger.logger_level = Level(result_ptr.level)

            finally:
                Logger._lib.free_log_result(result_ptr)
        else:
            raise LoggerError("Logger init received a null pointer")

    @classmethod
    def init(cls, level: Optional[Level] = None, file_name: Optional[str] = None):
        """
        Initialize a logger if it wasn't initialized before - this method is meant to be used when there is no intention to
        replace an existing logger. Otherwise, use `set_logger_config` for overriding the existing logger configs.
        The logger will filter all logs with a level lower than the given level.
        If given a file_name argument, will write the logs to files postfixed with file_name. If file_name isn't provided,
        the logs will be written to the console.

        Args:
            level (Optional[Level]): Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF].
                If log level isn't provided, the logger will be configured with default configuration.
                To turn off logging completely, set the level to Level.OFF.
            file_name (Optional[str]): If provided the target of the logs will be the file mentioned.
                Otherwise, logs will be printed to the console.
        """
        if cls._instance is None:
            cls._instance = cls(level, file_name)

    @classmethod
    def log(
        cls,
        log_level: Level,
        log_identifier: str,
        message: str,
        err: Optional[Exception] = None,
    ):
        """
        Logs the provided message if the provided log level is lower then the logger level.

        Args:
            log_level (Level): The log level of the provided message.
            log_identifier (str): The log identifier should give the log a context.
            message (str): The message to log.
            err (Optional[Exception]): The exception or error to log.
        """
        if not cls._instance:
            cls._instance = cls(None)
        if log_level.value > Logger.logger_level.value:
            return
        if err:
            message = f"{message}: {traceback.format_exception(type(err), err, err.__traceback__)}"
        c_identifier = Logger._ffi.new("char[]", log_identifier.encode(ENCODING))
        c_message = Logger._ffi.new("char[]", message.encode(ENCODING))
        result_ptr = Logger._lib.glide_log(log_level.value, c_identifier, c_message)

        if result_ptr != Logger._ffi.NULL:
            try:
                if result_ptr.log_error != Logger._ffi.NULL:
                    error_str = cast(
                        bytes, Logger._ffi.string(result_ptr.log_error)
                    ).decode(ENCODING)

                    # If the log failed due to invalid provided identifier or message,
                    # Log the FFI log error using logger_core directly
                    error_identifier = Logger._ffi.new(
                        "char[]", "Logger".encode(ENCODING)
                    )
                    error_message = Logger._ffi.new(
                        "char[]", f"log error: {error_str}".encode(ENCODING)
                    )
                    error_result_ptr = Logger._lib.glide_log(
                        Level.ERROR.value, error_identifier, error_message
                    )
                    if error_result_ptr != Logger._ffi.NULL:
                        Logger._lib.free_log_result(error_result_ptr)

            finally:
                Logger._lib.free_log_result(result_ptr)

        else:
            # Log the null pointer error using logger_core directly
            error_identifier = Logger._ffi.new("char[]", "Logger".encode(ENCODING))
            error_message = Logger._ffi.new(
                "char[]", "Log function returned a null pointer".encode(ENCODING)
            )
            error_result_ptr = Logger._lib.glide_log(
                Level.ERROR.value, error_identifier, error_message
            )
            if error_result_ptr != Logger._ffi.NULL:
                Logger._lib.free_log_result(error_result_ptr)

    @classmethod
    def set_logger_config(
        cls, level: Optional[Level] = None, file_name: Optional[str] = None
    ):
        """
        Creates a new logger instance and configure it with the provided log level and file name.

        Args:
            level (Optional[Level]): Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF].
                If log level isn't provided, the logger will be configured with default configuration.
                To turn off logging completely, set the level to OFF.
            file_name (Optional[str]): If provided the target of the logs will be the file mentioned.
                Otherwise, logs will be printed to the console.
        """
        Logger._instance = cls(level, file_name)

init(level=None, file_name=None) classmethod

Initialize a logger if it wasn't initialized before - this method is meant to be used when there is no intention to replace an existing logger. Otherwise, use set_logger_config for overriding the existing logger configs. The logger will filter all logs with a level lower than the given level. If given a file_name argument, will write the logs to files postfixed with file_name. If file_name isn't provided, the logs will be written to the console.

Parameters:

Name Type Description Default
level Optional[Level]

Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF]. If log level isn't provided, the logger will be configured with default configuration. To turn off logging completely, set the level to Level.OFF.

 None
file_name Optional[str]

If provided the target of the logs will be the file mentioned. Otherwise, logs will be printed to the console.

 None
Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/logger.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@classmethod
def init(cls, level: Optional[Level] = None, file_name: Optional[str] = None):
    """
    Initialize a logger if it wasn't initialized before - this method is meant to be used when there is no intention to
    replace an existing logger. Otherwise, use `set_logger_config` for overriding the existing logger configs.
    The logger will filter all logs with a level lower than the given level.
    If given a file_name argument, will write the logs to files postfixed with file_name. If file_name isn't provided,
    the logs will be written to the console.

    Args:
        level (Optional[Level]): Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF].
            If log level isn't provided, the logger will be configured with default configuration.
            To turn off logging completely, set the level to Level.OFF.
        file_name (Optional[str]): If provided the target of the logs will be the file mentioned.
            Otherwise, logs will be printed to the console.
    """
    if cls._instance is None:
        cls._instance = cls(level, file_name)

log(log_level, log_identifier, message, err=None) classmethod

Logs the provided message if the provided log level is lower then the logger level.

Parameters:

Name Type Description Default
log_level Level

The log level of the provided message.

 required
log_identifier str

The log identifier should give the log a context.

 required
message str

The message to log.

 required
err Optional[Exception]

The exception or error to log.

 None
Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/logger.py 
 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
@classmethod
def log(
    cls,
    log_level: Level,
    log_identifier: str,
    message: str,
    err: Optional[Exception] = None,
):
    """
    Logs the provided message if the provided log level is lower then the logger level.

    Args:
        log_level (Level): The log level of the provided message.
        log_identifier (str): The log identifier should give the log a context.
        message (str): The message to log.
        err (Optional[Exception]): The exception or error to log.
    """
    if not cls._instance:
        cls._instance = cls(None)
    if log_level.value > Logger.logger_level.value:
        return
    if err:
        message = f"{message}: {traceback.format_exception(type(err), err, err.__traceback__)}"
    c_identifier = Logger._ffi.new("char[]", log_identifier.encode(ENCODING))
    c_message = Logger._ffi.new("char[]", message.encode(ENCODING))
    result_ptr = Logger._lib.glide_log(log_level.value, c_identifier, c_message)

    if result_ptr != Logger._ffi.NULL:
        try:
            if result_ptr.log_error != Logger._ffi.NULL:
                error_str = cast(
                    bytes, Logger._ffi.string(result_ptr.log_error)
                ).decode(ENCODING)

                # If the log failed due to invalid provided identifier or message,
                # Log the FFI log error using logger_core directly
                error_identifier = Logger._ffi.new(
                    "char[]", "Logger".encode(ENCODING)
                )
                error_message = Logger._ffi.new(
                    "char[]", f"log error: {error_str}".encode(ENCODING)
                )
                error_result_ptr = Logger._lib.glide_log(
                    Level.ERROR.value, error_identifier, error_message
                )
                if error_result_ptr != Logger._ffi.NULL:
                    Logger._lib.free_log_result(error_result_ptr)

        finally:
            Logger._lib.free_log_result(result_ptr)

    else:
        # Log the null pointer error using logger_core directly
        error_identifier = Logger._ffi.new("char[]", "Logger".encode(ENCODING))
        error_message = Logger._ffi.new(
            "char[]", "Log function returned a null pointer".encode(ENCODING)
        )
        error_result_ptr = Logger._lib.glide_log(
            Level.ERROR.value, error_identifier, error_message
        )
        if error_result_ptr != Logger._ffi.NULL:
            Logger._lib.free_log_result(error_result_ptr)

set_logger_config(level=None, file_name=None) classmethod

Creates a new logger instance and configure it with the provided log level and file name.

Parameters:

Name Type Description Default
level Optional[Level]

Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF]. If log level isn't provided, the logger will be configured with default configuration. To turn off logging completely, set the level to OFF.

 None
file_name Optional[str]

If provided the target of the logs will be the file mentioned. Otherwise, logs will be printed to the console.

 None
Source code in doc-gen/valkey-glide/python/glide-sync/glide_sync/logger.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@classmethod
def set_logger_config(
    cls, level: Optional[Level] = None, file_name: Optional[str] = None
):
    """
    Creates a new logger instance and configure it with the provided log level and file name.

    Args:
        level (Optional[Level]): Set the logger level to one of [ERROR, WARN, INFO, DEBUG, TRACE, OFF].
            If log level isn't provided, the logger will be configured with default configuration.
            To turn off logging completely, set the level to OFF.
        file_name (Optional[str]): If provided the target of the logs will be the file mentioned.
            Otherwise, logs will be printed to the console.
    """
    Logger._instance = cls(level, file_name)