Skip to content

Code Documentation

Abhakliste

Abhakliste is a minimal task runner that prints a list of tasks and their status.

The object provides several methods such as run_context() and run_cmd() that can be used to run tasks in sequence. The status of each task is printed to the console. If a task fails, the error message and its traceback are printed to the console but the program continues. Otherwise all output to stdout and stderr is suppressed. At the end of the program, the object can be used to raise an exception if any of the tasks failed. This can be used to fail a CI build if any of the tasks failed.

Example 
from time import sleep
from abhakliste import Abhakliste

# Create a new Abhakliste instance.
abhaker = Abhakliste()

# Run first task in context
with abhaker.run_context(desc="Test 1"):
    sleep(1)

# Run second task as cmd
abhaker.run_cmd(["ls", "-l"], desc="Test 2"):

def _test3():
    sleep(1)
abhaker.run_function(_test3, desc="Test 3")

Attributes:

Name Type Description
error_runs int

Number of tasks that failed.
total_runs int

Number of tasks that were run.
Source code in abhakliste/abhakliste.py 
 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
 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
class Abhakliste:
    """Abhakliste is a minimal task runner that prints a list of tasks and their status.

    The object provides several methods such as
    [run_context()][abhakliste.abhakliste.Abhakliste.run_context] and
    [run_cmd()][abhakliste.abhakliste.Abhakliste.run_cmd] that can be used
    to run tasks in sequence.
    The status of each task is printed to the console.
    If a task fails, the error message and its traceback are printed to the console but the program
    continues.
    Otherwise all output to stdout and stderr is suppressed.
    At the end of the program, the object can be used to raise an exception if any
    of the tasks failed.
    This can be used to fail a CI build if any of the tasks failed.

    Example:
        ```python
        from time import sleep
        from abhakliste import Abhakliste

        # Create a new Abhakliste instance.
        abhaker = Abhakliste()

        # Run first task in context
        with abhaker.run_context(desc="Test 1"):
            sleep(1)

        # Run second task as cmd
        abhaker.run_cmd(["ls", "-l"], desc="Test 2"):

        def _test3():
            sleep(1)
        abhaker.run_function(_test3, desc="Test 3")
        ```

    Attributes:
        error_runs: Number of tasks that failed.
        total_runs: Number of tasks that were run.
    """

    def __init__(self) -> None:
        self.error_runs: int = 0
        self.total_runs: int = 0

    @contextmanager
    def run_context(self, desc: str) -> Generator[None, None, None]:
        """The context manager that is used to run a tasks.

        It provides a context manager in which the a single task is run and its status is printed.
        The following functionality is provided by the the context manager:

        - Print the description of the task.
        - Capture stdout and stderr.
        - Print the status of the task.
        - Print the error message and traceback if the task failed.

        Args:
            desc: A short description of the task.

        Yields:
            Nothing is yielded.

        Example:
            ```python
            from time import sleep
            from abhakliste import Abhakliste

            # Run first task in context
            abhaker = Abhakliste()
            with abhaker.run_context(desc="Test 1"):
                sleep(1)
            ```
        """
        func_err: Optional[Exception] = None

        self.total_runs += 1

        # print run description
        just_width = min(max(_get_terminal_width() - 2, 30), 100)
        print(desc.ljust(just_width, "."), end="")
        sys.stdout.flush()

        # run function and capture output
        with Capturing() as capture:
            try:
                yield
            except Exception as e:
                self.error_runs += 1
                func_err = e

        # print output
        if func_err is None:
            print("✅")
        else:
            print("🚨")

            # print caputred output
            if capture.stdout:
                print(f"{Colors.BLUE}{Colors.BOLD}==>{Colors.END} Stdout:")
                print(capture.stdout)
            if capture.stderr:
                print(f"{Colors.BLUE}{Colors.BOLD}==>{Colors.END} Stderr:")
                print(capture.stderr)

            # print traceback
            print(f"{Colors.BLUE}{Colors.BOLD}==>{Colors.END} Traceback:")
            print(traceback.format_exc())

    def run_function(self, func: Callable, desc: str, *args: Any, **kwargs: Any) -> None:
        """Run a function as a task and print its status.

        This is a helper method that runs a function in a context managed by `run_context()`.
        The function is called with the provided arguments.
        If the function raises an exception, the task is considered to have failed.
        See [run_context()][abhakliste.abhakliste.Abhakliste.run_context] for more information.

        Args:
            func: Function that will be run as a task.
            desc: A short description of the task.
            *args: Positional arguments that will be passed to the function.
            **kwargs: Keyword arguments that will be passed to the function.

        Example:
            ```python
            from time import sleep
            from abhakliste import Abhakliste

            # Define a test function
            def test_func(time: int) -> None:
                sleep(time)

            # Run function as task
            abhaker = Abhakliste()
            abhaker.run_function(test_func, desc="Test 1", time=1)
            ```
        """
        with self.run_context(desc):
            func(*args, **kwargs)

    def run_cmd(
        self,
        args: Union[str, List[Any]],
        desc: str,
        shell: bool = False,
        cwd: Optional[str] = None,
        text: Optional[Any] = None,
        **kwargs: Any,
    ) -> None:
        """Run a command line command as a task and print its status.

        This is a thin wrapper around `subprocess.Popen()` that runs a command line command,
        waits for it to finish and checks its return code.
        If the return code is not `0`, the command is considered to have failed.
        A description of the command and its status is printed to the console.

        Args:
            args: A string, or a sequence of program arguments.
            desc: A short description of the task.
            shell: If `True`, the command will be executed through the shell.
            cwd: Sets the current directory before the child is executed.
            text: If `True`, the `stdout` and `stderr` arguments must be `str` and will be
            **kwargs: Keyword arguments that will be passed to `subprocess.Popen()`.

        Raises:
            RuntimeError: The command failed and the return code was not `0`.

        Example:
            ```python
            from time import sleep
            from abhakliste import Abhakliste

            # Run command as task
            abhaker = Abhakliste()
            abhaker.run_cmd(["ls", "-l"], desc="Run ls"):
            ```
        """
        with self.run_context(desc):
            p = Popen(
                args=args,
                shell=shell,
                cwd=cwd,
                stdout=sys.stdout,
                stderr=sys.stderr,
                text=text,
                **kwargs,
            )
            p.wait()
            if p.returncode != 0:
                raise RuntimeError("The command exited with a non-zero exit code.")

    def raise_on_fail(self) -> None:
        """Raise an exception if any of the tasks failed.

        This method can be used to fail a CI build if any of the tasks failed or to raise an
        exception if the program is run from the command line.

        Raises:
            RuntimeError: At least one task failed.
        """
        if self.error_runs > 0:
            raise RuntimeError(f"{self.error_runs} of {self.total_runs} runs failed.")

raise_on_fail()

Raise an exception if any of the tasks failed.

This method can be used to fail a CI build if any of the tasks failed or to raise an exception if the program is run from the command line.

Raises:

Type Description
RuntimeError

At least one task failed.
Source code in abhakliste/abhakliste.py 
202
203
204
205
206
207
208
209
210
211
212
def raise_on_fail(self) -> None:
    """Raise an exception if any of the tasks failed.

    This method can be used to fail a CI build if any of the tasks failed or to raise an
    exception if the program is run from the command line.

    Raises:
        RuntimeError: At least one task failed.
    """
    if self.error_runs > 0:
        raise RuntimeError(f"{self.error_runs} of {self.total_runs} runs failed.")

run_cmd(args, desc, shell=False, cwd=None, text=None, **kwargs)

Run a command line command as a task and print its status.

This is a thin wrapper around subprocess.Popen() that runs a command line command, waits for it to finish and checks its return code. If the return code is not 0, the command is considered to have failed. A description of the command and its status is printed to the console.

Parameters:

Name Type Description Default
args Union[str, List[Any]]

A string, or a sequence of program arguments.

 required
desc str

A short description of the task.

 required
shell bool

If True, the command will be executed through the shell.

 False
cwd Optional[str]

Sets the current directory before the child is executed.

 None
text Optional[Any]

If True, the stdout and stderr arguments must be str and will be

 None
**kwargs Any

Keyword arguments that will be passed to subprocess.Popen().

 {}

Raises:

Type Description
RuntimeError

The command failed and the return code was not 0.
Example 
from time import sleep
from abhakliste import Abhakliste

# Run command as task
abhaker = Abhakliste()
abhaker.run_cmd(["ls", "-l"], desc="Run ls"):
Source code in abhakliste/abhakliste.py 
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
def run_cmd(
    self,
    args: Union[str, List[Any]],
    desc: str,
    shell: bool = False,
    cwd: Optional[str] = None,
    text: Optional[Any] = None,
    **kwargs: Any,
) -> None:
    """Run a command line command as a task and print its status.

    This is a thin wrapper around `subprocess.Popen()` that runs a command line command,
    waits for it to finish and checks its return code.
    If the return code is not `0`, the command is considered to have failed.
    A description of the command and its status is printed to the console.

    Args:
        args: A string, or a sequence of program arguments.
        desc: A short description of the task.
        shell: If `True`, the command will be executed through the shell.
        cwd: Sets the current directory before the child is executed.
        text: If `True`, the `stdout` and `stderr` arguments must be `str` and will be
        **kwargs: Keyword arguments that will be passed to `subprocess.Popen()`.

    Raises:
        RuntimeError: The command failed and the return code was not `0`.

    Example:
        ```python
        from time import sleep
        from abhakliste import Abhakliste

        # Run command as task
        abhaker = Abhakliste()
        abhaker.run_cmd(["ls", "-l"], desc="Run ls"):
        ```
    """
    with self.run_context(desc):
        p = Popen(
            args=args,
            shell=shell,
            cwd=cwd,
            stdout=sys.stdout,
            stderr=sys.stderr,
            text=text,
            **kwargs,
        )
        p.wait()
        if p.returncode != 0:
            raise RuntimeError("The command exited with a non-zero exit code.")

run_context(desc)

The context manager that is used to run a tasks.

It provides a context manager in which the a single task is run and its status is printed. The following functionality is provided by the the context manager:

  • Print the description of the task.
  • Capture stdout and stderr.
  • Print the status of the task.
  • Print the error message and traceback if the task failed.

Parameters:

Name Type Description Default
desc str

A short description of the task.

 required

Yields:

Type Description
Generator[None, None, None]

Nothing is yielded.
Example 
from time import sleep
from abhakliste import Abhakliste

# Run first task in context
abhaker = Abhakliste()
with abhaker.run_context(desc="Test 1"):
    sleep(1)
Source code in abhakliste/abhakliste.py 
 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
@contextmanager
def run_context(self, desc: str) -> Generator[None, None, None]:
    """The context manager that is used to run a tasks.

    It provides a context manager in which the a single task is run and its status is printed.
    The following functionality is provided by the the context manager:

    - Print the description of the task.
    - Capture stdout and stderr.
    - Print the status of the task.
    - Print the error message and traceback if the task failed.

    Args:
        desc: A short description of the task.

    Yields:
        Nothing is yielded.

    Example:
        ```python
        from time import sleep
        from abhakliste import Abhakliste

        # Run first task in context
        abhaker = Abhakliste()
        with abhaker.run_context(desc="Test 1"):
            sleep(1)
        ```
    """
    func_err: Optional[Exception] = None

    self.total_runs += 1

    # print run description
    just_width = min(max(_get_terminal_width() - 2, 30), 100)
    print(desc.ljust(just_width, "."), end="")
    sys.stdout.flush()

    # run function and capture output
    with Capturing() as capture:
        try:
            yield
        except Exception as e:
            self.error_runs += 1
            func_err = e

    # print output
    if func_err is None:
        print("✅")
    else:
        print("🚨")

        # print caputred output
        if capture.stdout:
            print(f"{Colors.BLUE}{Colors.BOLD}==>{Colors.END} Stdout:")
            print(capture.stdout)
        if capture.stderr:
            print(f"{Colors.BLUE}{Colors.BOLD}==>{Colors.END} Stderr:")
            print(capture.stderr)

        # print traceback
        print(f"{Colors.BLUE}{Colors.BOLD}==>{Colors.END} Traceback:")
        print(traceback.format_exc())

run_function(func, desc, *args, **kwargs)

Run a function as a task and print its status.

This is a helper method that runs a function in a context managed by run_context(). The function is called with the provided arguments. If the function raises an exception, the task is considered to have failed. See run_context() for more information.

Parameters:

Name Type Description Default
func Callable

Function that will be run as a task.

 required
desc str

A short description of the task.

 required
*args Any

Positional arguments that will be passed to the function.

 ()
**kwargs Any

Keyword arguments that will be passed to the function.

 {}
Example 
from time import sleep
from abhakliste import Abhakliste

# Define a test function
def test_func(time: int) -> None:
    sleep(time)

# Run function as task
abhaker = Abhakliste()
abhaker.run_function(test_func, desc="Test 1", time=1)
Source code in abhakliste/abhakliste.py 
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
def run_function(self, func: Callable, desc: str, *args: Any, **kwargs: Any) -> None:
    """Run a function as a task and print its status.

    This is a helper method that runs a function in a context managed by `run_context()`.
    The function is called with the provided arguments.
    If the function raises an exception, the task is considered to have failed.
    See [run_context()][abhakliste.abhakliste.Abhakliste.run_context] for more information.

    Args:
        func: Function that will be run as a task.
        desc: A short description of the task.
        *args: Positional arguments that will be passed to the function.
        **kwargs: Keyword arguments that will be passed to the function.

    Example:
        ```python
        from time import sleep
        from abhakliste import Abhakliste

        # Define a test function
        def test_func(time: int) -> None:
            sleep(time)

        # Run function as task
        abhaker = Abhakliste()
        abhaker.run_function(test_func, desc="Test 1", time=1)
        ```
    """
    with self.run_context(desc):
        func(*args, **kwargs)