ui: use ui.rich_text instead of rich.text.Text (#6690)

dvc/command/experiments.py

@@ -20,9 +20,8 @@ from dvc.utils.flatten import flatten
 from dvc.utils.serialize import encode_exception
 
 if TYPE_CHECKING:
-    from rich.text import Text
-
     from dvc.compare import TabularData
+    from dvc.ui import RichText
 
 
 logger = logging.getLogger(__name__)
@@ -292,8 +291,6 @@ def _format_time(datetime_obj, fill_value=FILL_VALUE, iso=False):
 
 
 def _extend_row(row, names, items, precision, fill_value=FILL_VALUE):
-    from rich.text import Text
-
     from dvc.compare import _format_field, with_value
 
     if not items:
@@ -308,10 +305,10 @@ def _extend_row(row, names, items, precision, fill_value=FILL_VALUE):
                 item.get(name),
                 FILL_VALUE_ERRORED if data.get("error", None) else fill_value,
             )
-            # wrap field data in rich.Text, otherwise rich may
+            # wrap field data in ui.rich_text, otherwise rich may
             # interpret unescaped braces from list/dict types as rich
             # markup tags
-            row.append(Text(str(_format_field(value, precision))))
+            row.append(ui.rich_text(str(_format_field(value, precision))))
 
 
 def _parse_filter_list(param_list):
@@ -362,17 +359,15 @@ def experiments_table(
     return td
 
 
-def prepare_exp_id(kwargs) -> "Text":
-    from rich.text import Text
-
+def prepare_exp_id(kwargs) -> "RichText":
     exp_name = kwargs["Experiment"]
     rev = kwargs["rev"]
     typ = kwargs.get("typ", "baseline")
 
     if typ == "baseline" or not exp_name:
-        text = Text(exp_name or rev)
+        text = ui.rich_text(exp_name or rev)
     else:
-        text = Text.assemble(rev, " [", (exp_name, "bold"), "]")
+        text = ui.rich_text.assemble(rev, " [", (exp_name, "bold"), "]")
 
     parent = kwargs.get("parent")
     suff = f" ({parent})" if parent else ""
@@ -380,7 +375,7 @@ def prepare_exp_id(kwargs) -> "Text":
 
     tree = experiment_types[typ]
     pref = f"{tree} " if tree else ""
-    return Text(pref) + text
+    return ui.rich_text(pref) + text
 
 
 def baseline_styler(typ):

dvc/ui/__init__.py

@@ -7,6 +7,7 @@ from typing import (
     Optional,
     Sequence,
     TextIO,
+    Type,
     Union,
 )
 
@@ -14,6 +15,7 @@ from funcy import cached_property
 
 if TYPE_CHECKING:
     from rich.status import Status
+    from rich.text import Text as RichText
 
     from dvc.progress import Tqdm
     from dvc.ui.table import Headers, Styles, TableData
@@ -103,6 +105,12 @@ class Console:
             )
             return print(*values, sep=sep, end=end, file=file)
 
+    @property
+    def rich_text(self) -> "Type[RichText]":
+        from rich.text import Text
+
+        return Text
+
     @staticmethod
     def progress(*args, **kwargs) -> "Tqdm":
         from dvc.progress import Tqdm

dvc/ui/table.py

@@ -8,15 +8,14 @@ from dvc.types import DictStrAny
 if TYPE_CHECKING:
     from rich.console import Console as RichConsole
     from rich.table import Table
-    from rich.text import Text
 
-    from dvc.ui import Console
+    from dvc.ui import Console, RichText
 
 
 SHOW_MAX_WIDTH = 1024
 
 
-CellT = Union[str, "Text"]  # Text is mostly compatible with str
+CellT = Union[str, "RichText"]  # RichText is mostly compatible with str
 Row = Sequence[CellT]
 TableData = Sequence[Row]
 Headers = Sequence[str]

dvc/updater.py

@@ -10,7 +10,7 @@ from dvc import __version__
 from dvc.utils.pkg import PKG
 
 if TYPE_CHECKING:
-    from rich.text import Text
+    from dvc.ui import RichText
 
 logger = logging.getLogger(__name__)
 
@@ -123,20 +123,23 @@ class Updater:
         current: str = None,
         color: str = "yellow",
         pkg: Optional[str] = None,
-    ) -> "Text":
-        from rich.text import Text
+    ) -> "RichText":
+        from dvc.ui import ui
 
         current = current or self.current
-        update_message = Text.from_markup(
+        update_message = ui.rich_text.from_markup(
             f"You are using dvc version [bold]{current}[/]; "
             f"however, version [bold]{latest}[/] is available."
         )
-        instruction = Text.from_markup(self._get_update_instructions(pkg=pkg))
-        return Text.assemble(
+        instruction = ui.rich_text.from_markup(
+            self._get_update_instructions(pkg=pkg)
+        )
+        return ui.rich_text.assemble(
             "\n", update_message, "\n", instruction, style=color
         )
 
-    def _get_update_instructions(self, pkg: Optional[str] = None) -> str:
+    @staticmethod
+    def _get_update_instructions(pkg: Optional[str] = None) -> str:
         if pkg in ("osxpkg", "exe", "binary"):
             return (
                 "To upgrade, uninstall dvc and reinstall from "

dvc/utils/strictyaml.py

@@ -9,6 +9,7 @@ Not to be confused with strictyaml, a python library with similar motivations.
 from typing import TYPE_CHECKING, Any, Callable, List, TypeVar
 
 from dvc.exceptions import DvcException, PrettyDvcException
+from dvc.ui import ui
 from dvc.utils.serialize import (
     EncodingError,
     YAMLFileCorruptedError,
@@ -17,19 +18,17 @@ from dvc.utils.serialize import (
 
 if TYPE_CHECKING:
     from rich.syntax import Syntax
-    from rich.text import Text
     from ruamel.yaml import StreamMark
 
     from dvc.fs.base import BaseFileSystem
+    from dvc.ui import RichText
 
 
 _T = TypeVar("_T")
 
 
-def _prepare_cause(cause: str) -> "Text":
-    from rich.text import Text
-
-    return Text(cause, style="bold")
+def _prepare_cause(cause: str) -> "RichText":
+    return ui.rich_text(cause, style="bold")
 
 
 def _prepare_code_snippets(code: str, start_line: int = 1) -> "Syntax":
@@ -56,7 +55,6 @@ class YAMLSyntaxError(PrettyDvcException, YAMLFileCorruptedError):
     def __pretty_exc__(self, **kwargs: Any) -> None:
         from ruamel.yaml.error import MarkedYAMLError
 
-        from dvc.ui import ui
         from dvc.utils import relpath
 
         exc = self.exc.__cause__
@@ -69,7 +67,9 @@ class YAMLSyntaxError(PrettyDvcException, YAMLFileCorruptedError):
         def prepare_linecol(mark: "StreamMark") -> str:
             return f"in line {mark.line + 1}, column {mark.column + 1}"
 
-        def prepare_message(message: str, mark: "StreamMark" = None) -> "Text":
+        def prepare_message(
+            message: str, mark: "StreamMark" = None
+        ) -> "RichText":
             cause = ", ".join(
                 [message.capitalize(), prepare_linecol(mark) if mark else ""]
             )