Skip to content

log

EnvVars

Bases: BaseModel

Class to store the values of the environment variables.

Attributes:

Name Type Description
ci_env bool

...
github_token str

...
server_url str

...
run_id str

...
run_number str

...
run_url str

...
git_branch str

...
git_sha str

...
triggered_by str

...
triggered_by_email str

...
triggered_by_user str

...
Source code in m/log/ci_tools/types.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class EnvVars(BaseModel):
    """Class to store the values of the environment variables."""

    # pylint: disable=too-many-instance-attributes
    ci_env: bool = False
    github_token: str = ''
    server_url: str = ''
    run_id: str = ''
    run_number: str = ''
    run_url: str = ''
    git_branch: str = ''
    git_sha: str = ''
    triggered_by: str = ''
    triggered_by_email: str = ''
    triggered_by_user: str = ''

Logger

Wrapper for python Logger to enable inserting logs in fp loops.

This is also so that we may attach context data for a message.

We can access the actual logger by using inst.

Source code in m/log/logger.py 
 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
class Logger:  # noqa: WPS230 - loggers have many attributes
    """Wrapper for python Logger to enable inserting logs in fp loops.

    This is also so that we may attach context data for a message.

    We can access the actual logger by using `inst`.
    """

    # pylint: disable=too-many-instance-attributes
    def __init__(self, name: str):
        """Initialize the Logger.

        Args:
            name: The name of a logger.
        """
        self.inst = logging.getLogger(name)
        self.debug = partial(log_func_wrapper, self.inst.debug)
        self.info = partial(log_func_wrapper, self.inst.info)  # noqa: WPS110
        self.warning = partial(log_func_wrapper, self.inst.warning)
        self.exception = partial(log_func_wrapper, self.inst.exception)
        self.critical = partial(log_func_wrapper, self.inst.critical)
        self.log = partial(log_func_wrapper, self.inst.log)
        self.prompt = partial(
            log_func_wrapper,
            cast(PromptLogger, self.inst).prompt,
        )
        self.error = partial(log_func_wrapper, self.inst.error)

    def open_block(
        self,
        name: str,
        description: str,
        stderr: bool = False,
    ) -> Res[int]:
        """Group log lines.

        Signals the formatter that the next log lines should be placed in a
        group.

        Args:
            name: The name of the block to open.
            description: Not supported.
            stderr: Force message to be displayed in stderr.

        Returns:
            A OneOf containing 0 to make it easier to call in fp for loops.
        """
        func = self.inst.error if stderr else self.inst.info
        func(
            'OPEN_BLOCK',
            extra={'open_block': (name, description)},
            stacklevel=2,
        )
        return Good(0)

    def close_block(
        self,
        name: str,
        stderr: bool = False,
    ) -> Res[int]:
        """Close a group log lines.

        Signals the formatter that the current group of lines should end.

        Args:
            name: The name of the block to close.
            stderr: Force message to be displayed in stderr.

        Returns:
            A OneOf containing 0 to make it easier to call in fp for loops.
        """
        func = self.inst.error if stderr else self.inst.info
        func(
            'CLOSE_BLOCK',
            extra={'close_block': name},
            stacklevel=2,
        )
        return Good(0)

    def error_block(
        self,
        msg: str | Message,
        context: dict | Issue,
    ) -> Res[int]:
        """Display an error block.

        Args:
            msg: The error message to display.
            context: The dict/Issue to display in a block.

        Returns:
            A OneOf containing 0 to make it easier to call in fp for loops.
        """
        self.error(msg)
        self.open_block('error', '', stderr=True)
        self.error('', context)
        self.close_block('error', stderr=True)
        return Good(0)

    def waning_block(
        self,
        msg: str | Message,
        context: dict | Issue,
    ) -> Res[int]:
        """Display a warning block.

        Args:
            msg: The warning message to display.
            context: The dict/Issue to display in a block.

        Returns:
            A OneOf containing 0 to make it easier to call in fp for loops.
        """
        self.warning(msg)
        self.open_block('warning', '', stderr=True)
        self.warning('', context)
        self.close_block('warning', stderr=True)
        return Good(0)

__init__(name)

Initialize the Logger.

Parameters:

Name Type Description Default
name str

The name of a logger.

 required
Source code in m/log/logger.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def __init__(self, name: str):
    """Initialize the Logger.

    Args:
        name: The name of a logger.
    """
    self.inst = logging.getLogger(name)
    self.debug = partial(log_func_wrapper, self.inst.debug)
    self.info = partial(log_func_wrapper, self.inst.info)  # noqa: WPS110
    self.warning = partial(log_func_wrapper, self.inst.warning)
    self.exception = partial(log_func_wrapper, self.inst.exception)
    self.critical = partial(log_func_wrapper, self.inst.critical)
    self.log = partial(log_func_wrapper, self.inst.log)
    self.prompt = partial(
        log_func_wrapper,
        cast(PromptLogger, self.inst).prompt,
    )
    self.error = partial(log_func_wrapper, self.inst.error)

close_block(name, stderr=False)

Close a group log lines.

Signals the formatter that the current group of lines should end.

Parameters:

Name Type Description Default
name str

The name of the block to close.

 required
stderr bool

Force message to be displayed in stderr.

 False

Returns:

Type Description
Res[int]

A OneOf containing 0 to make it easier to call in fp for loops.
Source code in m/log/logger.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def close_block(
    self,
    name: str,
    stderr: bool = False,
) -> Res[int]:
    """Close a group log lines.

    Signals the formatter that the current group of lines should end.

    Args:
        name: The name of the block to close.
        stderr: Force message to be displayed in stderr.

    Returns:
        A OneOf containing 0 to make it easier to call in fp for loops.
    """
    func = self.inst.error if stderr else self.inst.info
    func(
        'CLOSE_BLOCK',
        extra={'close_block': name},
        stacklevel=2,
    )
    return Good(0)

error_block(msg, context)

Display an error block.

Parameters:

Name Type Description Default
msg str | Message

The error message to display.

 required
context dict | Issue

The dict/Issue to display in a block.

 required

Returns:

Type Description
Res[int]

A OneOf containing 0 to make it easier to call in fp for loops.
Source code in m/log/logger.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def error_block(
    self,
    msg: str | Message,
    context: dict | Issue,
) -> Res[int]:
    """Display an error block.

    Args:
        msg: The error message to display.
        context: The dict/Issue to display in a block.

    Returns:
        A OneOf containing 0 to make it easier to call in fp for loops.
    """
    self.error(msg)
    self.open_block('error', '', stderr=True)
    self.error('', context)
    self.close_block('error', stderr=True)
    return Good(0)

open_block(name, description, stderr=False)

Group log lines.

Signals the formatter that the next log lines should be placed in a group.

Parameters:

Name Type Description Default
name str

The name of the block to open.

 required
description str

Not supported.

 required
stderr bool

Force message to be displayed in stderr.

 False

Returns:

Type Description
Res[int]

A OneOf containing 0 to make it easier to call in fp for loops.
Source code in m/log/logger.py 
 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
def open_block(
    self,
    name: str,
    description: str,
    stderr: bool = False,
) -> Res[int]:
    """Group log lines.

    Signals the formatter that the next log lines should be placed in a
    group.

    Args:
        name: The name of the block to open.
        description: Not supported.
        stderr: Force message to be displayed in stderr.

    Returns:
        A OneOf containing 0 to make it easier to call in fp for loops.
    """
    func = self.inst.error if stderr else self.inst.info
    func(
        'OPEN_BLOCK',
        extra={'open_block': (name, description)},
        stacklevel=2,
    )
    return Good(0)

waning_block(msg, context)

Display a warning block.

Parameters:

Name Type Description Default
msg str | Message

The warning message to display.

 required
context dict | Issue

The dict/Issue to display in a block.

 required

Returns:

Type Description
Res[int]

A OneOf containing 0 to make it easier to call in fp for loops.
Source code in m/log/logger.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def waning_block(
    self,
    msg: str | Message,
    context: dict | Issue,
) -> Res[int]:
    """Display a warning block.

    Args:
        msg: The warning message to display.
        context: The dict/Issue to display in a block.

    Returns:
        A OneOf containing 0 to make it easier to call in fp for loops.
    """
    self.warning(msg)
    self.open_block('warning', '', stderr=True)
    self.warning('', context)
    self.close_block('warning', stderr=True)
    return Good(0)

Message

Bases: BaseModel

Parameters needed to deliver a warning or error message.

Attributes:

Name Type Description
msg str

message to display
title str | None

custom title
file str | None

filename
line str | None

line number, starting at 1
end_line str | None

end line number
col str | None

column number, starting at 1
end_col str | None

end column number
Source code in m/log/ci_tools/types.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class Message(BaseModel):
    """Parameters needed to deliver a warning or error message."""

    msg: str = Field(description='message to display')
    title: str | None = Field(default=None, description='custom title')
    file: str | None = Field(  # noqa: WPS110 - required by Github
        default=None,
        description='filename',
    )
    line: str | None = Field(
        default=None,
        description='line number, starting at 1',
    )
    end_line: str | None = Field(default=None, description='end line number')
    col: str | None = Field(
        default=None,
        description='column number, starting at 1',
    )
    end_col: str | None = Field(default=None, description='end column number')

    model_config = ConfigDict(arbitrary_types_allowed=True)

get_ci_tool()

Return the current CI Tool based on the environment variables.

Returns:

Type Description
ProviderModule

A ProviderModule instance with methods to provide messages in
ProviderModule

a CI environment.
Source code in m/log/ci_tools/ci_tools.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
def get_ci_tool() -> ProviderModule:
    """Return the current CI Tool based on the environment variables.

    Returns:
        A `ProviderModule` instance with methods to provide messages in
        a CI environment.
    """
    env = mio.env
    if env('GITHUB_ACTIONS'):
        return gh_tool
    if env('TC') or env('TEAMCITY'):
        return tc_tool
    return local_tool

logging_config(level=None, json_file='')

Apply a configuration to the logs.

Parameters:

Name Type Description Default
level int | None

The logging level, defaults to INFO.

 None
json_file str

Optional file name where to store each log record as json.

 ''
Source code in m/log/config.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def logging_config(level: int | None = None, json_file: str = '') -> None:
    """Apply a configuration to the logs.

    Args:
        level: The logging level, defaults to INFO.
        json_file: Optional file name where to store each log record as json.
    """
    formatter = CiFormatter()
    stdout_handler = StdOutHandler(formatter)
    stderr_handler = StdErrHandler(formatter)
    all_handlers = [stderr_handler, stdout_handler]
    if json_file:
        json_formatter = JsonFormatter()
        all_handlers.append(JsonFileHandler(json_formatter, json_file))

    debug_logs = mio.env('DEBUG_M_LOGS', 'false') == 'true'
    default_level = logging.DEBUG if debug_logs else logging.INFO
    logging_level = level if level is not None else default_level
    logging.basicConfig(
        level=logging_level,
        handlers=all_handlers,
        force=True,
    )