Skip to content

Context manager

This part of the project documentation focuses on an information-oriented approach. Use it as a reference for the technical implementation of waterfall-logging.context_manager.

waterfall 

waterfall(log: Waterfall, variable_names: List[str], markdown_kwargs: Optional[Dict] = None)

Waterfall context manager decorator.

Parameters:

Name Type Description Default
log Waterfall

Waterfall object

 required
variable_names List

array of variables names to log waterfall

 required
markdown_kwargs dict

keyword arguments to save the markdown with to_markdown(**markdown_kwargs)

 None

Examples:

>>> import pandas as pd
>>> import numpy as np
>>> from waterfall_logging.log import PandasWaterfall
>>> from waterfall_logging.context_manager import waterfall
>>> waterfall_log = PandasWaterfall(table_name='context_manager_example', columns=['col1'])
>>> @waterfall(log=waterfall_log, variable_names=['table'], markdown_kwargs={'buf': 'ctx_manager_example.md'})
... def main():
...     table = pd.DataFrame({'col1': [0, np.nan, 1, 2, 3], 'col2': [3, 4, np.nan, 5, 6]})
...     table.dropna(axis=0)
...
>>> main()
>>> print(waterfall_log.to_markdown())
| Table                   |   col1 |   Δ col1 |   Rows |   Δ Rows | Reason   | Configurations flag   |
|:------------------------|-------:|---------:|-------:|---------:|:---------|:----------------------|
| context_manager_example |      5 |        0 |      5 |        0 |          |                       |
>>> print("The `waterfall_log` is also saved in the file `ctx_manager_example.md`!")

Returns:

Name Type Description
decorator Callable

decorator object
Source code in waterfall_logging/context_manager.py 
 9
10
11
12
13
14
15
16
17
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
def waterfall(log: Waterfall, variable_names: List[str], markdown_kwargs: Optional[Dict] = None):
    """Waterfall context manager decorator.

    Args:
        log (Waterfall): Waterfall object
        variable_names (List): array of variables names to log waterfall
        markdown_kwargs (dict): keyword arguments to save the markdown with to_markdown(**markdown_kwargs)

    Examples:
        >>> import pandas as pd
        >>> import numpy as np
        >>> from waterfall_logging.log import PandasWaterfall
        >>> from waterfall_logging.context_manager import waterfall
        >>> waterfall_log = PandasWaterfall(table_name='context_manager_example', columns=['col1'])
        >>> @waterfall(log=waterfall_log, variable_names=['table'], markdown_kwargs={'buf': 'ctx_manager_example.md'})
        ... def main():
        ...     table = pd.DataFrame({'col1': [0, np.nan, 1, 2, 3], 'col2': [3, 4, np.nan, 5, 6]})
        ...     table.dropna(axis=0)
        ...
        >>> main()
        >>> print(waterfall_log.to_markdown())
        | Table                   |   col1 |   Δ col1 |   Rows |   Δ Rows | Reason   | Configurations flag   |
        |:------------------------|-------:|---------:|-------:|---------:|:---------|:----------------------|
        | context_manager_example |      5 |        0 |      5 |        0 |          |                       |
        >>> print("The `waterfall_log` is also saved in the file `ctx_manager_example.md`!")


    Returns:
        decorator (Callable): decorator object

    """
    markdown_kwargs = {} if not markdown_kwargs else markdown_kwargs

    def decorator_waterfall(method: Callable):
        """"""

        @functools.wraps(method)
        def wrapper_waterfall(*args, **kwargs):
            """"""
            with WaterfallContext(log, method.__name__, variable_names, markdown_kwargs):
                return_value = method(*args, **kwargs)
            return return_value

        return wrapper_waterfall

    return decorator_waterfall

WaterfallContext

Waterfall context to trace any function calls inside the context.

Source code in waterfall_logging/context_manager.py 
 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
class WaterfallContext:
    """Waterfall context to trace any function calls inside the context."""

    def __init__(self, log: Waterfall, name: str, variable_names: List[str], markdown_kwargs: Optional[Dict] = None):
        """

        Args:
            log (Waterfall): waterfall logging object
            name (str): name of the function that is to de debugged
            variable_names (List): array of variables names to log waterfall
            markdown_kwargs (dict): keyword arguments to save the markdown with to_markdown(**markdown_kwargs)

        """
        self.name = name
        self.variable_names = variable_names
        self.log = log
        self.markdown_kwargs = {} if not markdown_kwargs else markdown_kwargs

        self._df = None

    def __enter__(self):
        """Enter the trace."""
        sys.settrace(self.trace_calls)

    def __exit__(self, *args, **kwargs):
        """Exit the trace and export Waterfall object to Markdown."""
        self.log.to_markdown(**self.markdown_kwargs)
        sys.settrace = None

    def trace_calls(self, frame: types.FrameType, event, arg: Any):
        """

        Args:
            frame (types.FrameType): a tracing frame
            event (str): a tracing event
            arg (Any): a tracing argument

        Returns:
            traced_lines (types.MethodType): traced lines

        """
        if event != "call":
            # only trace our call to the decorated function
            return
        elif frame.f_code.co_name != self.name:
            # return the trace function to use when you go into that function call
            return

        traced_lines = self.trace_lines
        return traced_lines

    def trace_lines(self, frame: types.FrameType, event: str, arg: Any) -> None:
        """

        Args:
            frame (types.FrameType): a tracing frame
            event (str): a tracing event
            arg (Any): a tracing argument

        Returns:
            None

        """
        if event not in ["line", "return"]:
            return

        local_vars = frame.f_locals

        for var_name, var_value in local_vars.items():
            if var_name in self.variable_names:
                df = local_vars[var_name]
                if df is not self._df:
                    self.log.log(df)
                    self._df = df

__enter__ 

__enter__()

Enter the trace.

Source code in waterfall_logging/context_manager.py 
77
78
79
def __enter__(self):
    """Enter the trace."""
    sys.settrace(self.trace_calls)

__exit__ 

__exit__(*args, **kwargs)

Exit the trace and export Waterfall object to Markdown.

Source code in waterfall_logging/context_manager.py 
81
82
83
84
def __exit__(self, *args, **kwargs):
    """Exit the trace and export Waterfall object to Markdown."""
    self.log.to_markdown(**self.markdown_kwargs)
    sys.settrace = None

__init__ 

__init__(log: Waterfall, name: str, variable_names: List[str], markdown_kwargs: Optional[Dict] = None)

Parameters:

Name Type Description Default
log Waterfall

waterfall logging object

 required
name str

name of the function that is to de debugged

 required
variable_names List

array of variables names to log waterfall

 required
markdown_kwargs dict

keyword arguments to save the markdown with to_markdown(**markdown_kwargs)

 None
Source code in waterfall_logging/context_manager.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def __init__(self, log: Waterfall, name: str, variable_names: List[str], markdown_kwargs: Optional[Dict] = None):
    """

    Args:
        log (Waterfall): waterfall logging object
        name (str): name of the function that is to de debugged
        variable_names (List): array of variables names to log waterfall
        markdown_kwargs (dict): keyword arguments to save the markdown with to_markdown(**markdown_kwargs)

    """
    self.name = name
    self.variable_names = variable_names
    self.log = log
    self.markdown_kwargs = {} if not markdown_kwargs else markdown_kwargs

    self._df = None

trace_calls 

trace_calls(frame: types.FrameType, event, arg: Any)

Parameters:

Name Type Description Default
frame types.FrameType

a tracing frame

 required
event str

a tracing event

 required
arg Any

a tracing argument

 required

Returns:

Name Type Description
traced_lines types.MethodType

traced lines
Source code in waterfall_logging/context_manager.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def trace_calls(self, frame: types.FrameType, event, arg: Any):
    """

    Args:
        frame (types.FrameType): a tracing frame
        event (str): a tracing event
        arg (Any): a tracing argument

    Returns:
        traced_lines (types.MethodType): traced lines

    """
    if event != "call":
        # only trace our call to the decorated function
        return
    elif frame.f_code.co_name != self.name:
        # return the trace function to use when you go into that function call
        return

    traced_lines = self.trace_lines
    return traced_lines

trace_lines 

trace_lines(frame: types.FrameType, event: str, arg: Any) -> None

Parameters:

Name Type Description Default
frame types.FrameType

a tracing frame

 required
event str

a tracing event

 required
arg Any

a tracing argument

 required

Returns:

Type Description
None

None
Source code in waterfall_logging/context_manager.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def trace_lines(self, frame: types.FrameType, event: str, arg: Any) -> None:
    """

    Args:
        frame (types.FrameType): a tracing frame
        event (str): a tracing event
        arg (Any): a tracing argument

    Returns:
        None

    """
    if event not in ["line", "return"]:
        return

    local_vars = frame.f_locals

    for var_name, var_value in local_vars.items():
        if var_name in self.variable_names:
            df = local_vars[var_name]
            if df is not self._df:
                self.log.log(df)
                self._df = df