9903255773

Committed 12 Jul 2024 06:06AM UTC coverage: 77.595% (-8.5%) from 86.128%

Build # 9903255773

Build Type

Pull #3730

github

Committed by

RenkuBot

Commit Message

chore: release v2.9.4

Pull Request Pull Request #3730: chore: release v2.9.4

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

2624 existing lines in 126 files now uncovered.

23855 of 30743 relevant lines covered (77.59%)

3.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.33

/renku/ui/cli/utils/callback.py

# Copyright Swiss Data Science Center (SDSC). A partnership between
# École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Communicator class for printing click output."""

import sys
from contextlib import contextmanager

import click
from tqdm import tqdm
from yaspin import yaspin

import renku.ui.cli.utils.color as color
from renku.core.util.communication import CommunicationCallback


class StandardOutput(CommunicationCallback):
    """Communication listener that outputs to stdout/stderr."""

    def __init__(self):
        super().__init__()
        self._progress_bars = {}
        self._progress_types = ["download"]
        self._spinner = None

    @contextmanager
    def pause_spinner(self):
        """Pause Spinner so output gets actually written."""
        if self._spinner:
            self._spinner.hide()

        yield

        if self._spinner:
            self._spinner.show()

    def echo(self, msg, end="\n"):
        """Write a message."""
        with CommunicationCallback.lock, self.pause_spinner():
            print(msg, end=end)

    def info(self, msg):
        """Write an info message."""
        with CommunicationCallback.lock, self.pause_spinner():
            print(msg)

    def warn(self, msg):
        """Write a warning message."""
        with CommunicationCallback.lock, self.pause_spinner():
            print(msg)

    def error(self, msg):
        """Write an error message."""
        with CommunicationCallback.lock, self.pause_spinner():
            print(msg, file=sys.stderr)

    def confirm(self, msg, abort=False, warning=False, default=False):
        """Get confirmation for an action."""
        return False

    def start_progress(self, name, total, **kwargs):
        """Start a new tqdm progressbar."""
        if name in self._progress_bars:
            raise ValueError(f"Name {name} is already a registered progressbar.")

        if "type" not in kwargs:
            kwargs["type"] = "download"

        if kwargs["type"] not in self._progress_types:
            self._progress_bars[name] = None
        elif kwargs["type"] == "download":
            self._progress_bars[name] = tqdm(
                total=total,
                unit="iB",
                unit_scale=True,
                desc=name,
                leave=False,
                bar_format="{desc:.32}: {percentage:3.0f}%|{bar}{r_bar}",
            )

    def update_progress(self, name, amount):
        """Update a progressbar."""
        if name not in self._progress_bars or not self._progress_bars[name]:
            return

        self._progress_bars[name].update(amount)

    def finalize_progress(self, name):
        """End a progressbar."""
        if name not in self._progress_bars or not self._progress_bars[name]:
            return

        self._progress_bars[name].close()
        del self._progress_bars[name]

    @contextmanager
    def busy(self, msg):
        """Indicate busy status using a spinner."""
        self._spinner = yaspin(text=msg)
        try:
            self._spinner.start()
            yield
        finally:
            self._spinner.stop()
            self._spinner = None


class ClickCallback(StandardOutput):
    """CommunicationCallback implementation for ``click`` messages."""

    INFO = click.style("Info: ", bold=True, fg=color.BLUE)
    WARNING = click.style("Warning: ", bold=True, fg=color.YELLOW)
    ERROR = click.style("Error: ", bold=True, fg=color.RED)

    def echo(self, msg, end="\n"):
        """Write a message."""
        new_line = True
        if end != "\n":
            msg = msg + end
            new_line = False
        with self.pause_spinner():
            click.echo(msg, nl=new_line)

    def info(self, msg):
        """Write an info message."""
        with self.pause_spinner():
            click.echo(self.INFO + msg)

    def warn(self, msg):
        """Write a warning message."""
        with self.pause_spinner():
            click.echo(self.WARNING + msg)

    def error(self, msg):
        """Write an error message."""
        with self.pause_spinner():
            click.echo(self.ERROR + msg, err=True)

    def has_prompt(self):
        """Return True if communicator provides a direct prompt to users."""
        return True

    def confirm(self, msg, abort=False, warning=False, default=False):
        """Get confirmation for an action using a prompt."""
        prefix = self.WARNING if warning else ""
        with self.pause_spinner():
            return click.confirm(prefix + msg, abort=abort, default=default)

    def prompt(self, msg, type=None, default=None, **kwargs):
        """Show a message prompt from the first callback that has a prompt."""
        with self.pause_spinner():
            return click.prompt(msg, type=type, default=default, **kwargs)

1	# Copyright Swiss Data Science Center (SDSC). A partnership between
2	# École Polytechnique Fédérale de Lausanne (EPFL) and
3	# Eidgenössische Technische Hochschule Zürich (ETHZ).
4	#
5	# Licensed under the Apache License, Version 2.0 (the "License");
6	# you may not use this file except in compliance with the License.
7	# You may obtain a copy of the License at
8	#
9	# http://www.apache.org/licenses/LICENSE-2.0
10	#
11	# Unless required by applicable law or agreed to in writing, software
12	# distributed under the License is distributed on an "AS IS" BASIS,
13	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	# See the License for the specific language governing permissions and
15	# limitations under the License.
16	"""Communicator class for printing click output."""	7✔
17
18	import sys	7✔
19	from contextlib import contextmanager	7✔
20
21	import click	7✔
22	from tqdm import tqdm	7✔
23	from yaspin import yaspin	7✔
24
25	import renku.ui.cli.utils.color as color	7✔
26	from renku.core.util.communication import CommunicationCallback	7✔
27
28
29	class StandardOutput(CommunicationCallback):	7✔
30	"""Communication listener that outputs to stdout/stderr."""
31
32	def __init__(self):	7✔
33	super().__init__()	6✔
34	self._progress_bars = {}	6✔
35	self._progress_types = ["download"]	6✔
36	self._spinner = None	6✔
37
38	@contextmanager	7✔
39	def pause_spinner(self):	7✔
40	"""Pause Spinner so output gets actually written."""
41	if self._spinner:	3✔
UNCOV 42	self._spinner.hide()	×
43
44	yield	3✔
45
46	if self._spinner:	3✔
UNCOV 47	self._spinner.show()	×
48
49	def echo(self, msg, end="\n"):	7✔
50	"""Write a message."""
51	with CommunicationCallback.lock, self.pause_spinner():	×
52	print(msg, end=end)	×
53
54	def info(self, msg):	7✔
55	"""Write an info message."""
56	with CommunicationCallback.lock, self.pause_spinner():	×
57	print(msg)	×
58
59	def warn(self, msg):	7✔
60	"""Write a warning message."""
61	with CommunicationCallback.lock, self.pause_spinner():	×
62	print(msg)	×
63
64	def error(self, msg):	7✔
65	"""Write an error message."""
66	with CommunicationCallback.lock, self.pause_spinner():	×
67	print(msg, file=sys.stderr)	×
68
69	def confirm(self, msg, abort=False, warning=False, default=False):	7✔
70	"""Get confirmation for an action."""
71	return False	×
72
73	def start_progress(self, name, total, **kwargs):	7✔
74	"""Start a new tqdm progressbar."""
75	if name in self._progress_bars:	2✔
76	raise ValueError(f"Name {name} is already a registered progressbar.")	×
77
78	if "type" not in kwargs:	2✔
79	kwargs["type"] = "download"	2✔
80
81	if kwargs["type"] not in self._progress_types:	2✔
82	self._progress_bars[name] = None	×
83	elif kwargs["type"] == "download":	2✔
84	self._progress_bars[name] = tqdm(	2✔
85	total=total,
86	unit="iB",
87	unit_scale=True,
88	desc=name,
89	leave=False,
90	bar_format="{desc:.32}: {percentage:3.0f}%\|{bar}{r_bar}",
91	)
92
93	def update_progress(self, name, amount):	7✔
94	"""Update a progressbar."""
95	if name not in self._progress_bars or not self._progress_bars[name]:	2✔
96	return	1✔
97
98	self._progress_bars[name].update(amount)	2✔
99
100	def finalize_progress(self, name):	7✔
101	"""End a progressbar."""
102	if name not in self._progress_bars or not self._progress_bars[name]:	2✔
103	return	1✔
104
105	self._progress_bars[name].close()	2✔
106	del self._progress_bars[name]	2✔
107
108	@contextmanager	7✔
109	def busy(self, msg):	7✔
110	"""Indicate busy status using a spinner."""
111	self._spinner = yaspin(text=msg)	2✔
112	try:	2✔
113	self._spinner.start()	2✔
114	yield	2✔
115	finally:
116	self._spinner.stop()	2✔
117	self._spinner = None	2✔
118
119
120	class ClickCallback(StandardOutput):	7✔
121	"""CommunicationCallback implementation for ``click`` messages."""
122
123	INFO = click.style("Info: ", bold=True, fg=color.BLUE)	7✔
124	WARNING = click.style("Warning: ", bold=True, fg=color.YELLOW)	7✔
125	ERROR = click.style("Error: ", bold=True, fg=color.RED)	7✔
126
127	def echo(self, msg, end="\n"):	7✔
128	"""Write a message."""
129	new_line = True	3✔
130	if end != "\n":	3✔
131	msg = msg + end	2✔
132	new_line = False	2✔
133	with self.pause_spinner():	3✔
134	click.echo(msg, nl=new_line)	3✔
135
136	def info(self, msg):	7✔
137	"""Write an info message."""
138	with self.pause_spinner():	3✔
139	click.echo(self.INFO + msg)	3✔
140
141	def warn(self, msg):	7✔
142	"""Write a warning message."""
143	with self.pause_spinner():	3✔
144	click.echo(self.WARNING + msg)	3✔
145
146	def error(self, msg):	7✔
147	"""Write an error message."""
148	with self.pause_spinner():	×
149	click.echo(self.ERROR + msg, err=True)	×
150
151	def has_prompt(self):	7✔
152	"""Return True if communicator provides a direct prompt to users."""
153	return True	2✔
154
155	def confirm(self, msg, abort=False, warning=False, default=False):	7✔
156	"""Get confirmation for an action using a prompt."""
157	prefix = self.WARNING if warning else ""	1✔
158	with self.pause_spinner():	1✔
159	return click.confirm(prefix + msg, abort=abort, default=default)	1✔
160
161	def prompt(self, msg, type=None, default=None, **kwargs):	7✔
162	"""Show a message prompt from the first callback that has a prompt."""
163	with self.pause_spinner():	2✔
164	return click.prompt(msg, type=type, default=default, **kwargs)	2✔

SwissDataScienceCenter / renku-python / 9903255773

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous