Merge branch 'main' into docs/disable-preview

package/kedro_viz/data_access/managers.py

@@ -9,7 +9,6 @@
 from kedro.io import DataCatalog
 from kedro.pipeline import Pipeline as KedroPipeline
 from kedro.pipeline.node import Node as KedroNode
-from kedro.pipeline.pipeline import _strip_transcoding
 from sqlalchemy.orm import sessionmaker
 
 from kedro_viz.constants import DEFAULT_REGISTERED_PIPELINE_ID, ROOT_MODULAR_PIPELINE_ID
@@ -26,6 +25,7 @@
     TranscodedDataNode,
 )
 from kedro_viz.services import layers_services, modular_pipelines_services
+from kedro_viz.utils import _strip_transcoding
 
 from .repositories import (
     CatalogRepository,

package/kedro_viz/data_access/repositories/catalog.py

@@ -6,17 +6,10 @@
 from typing import TYPE_CHECKING, Dict, Optional
 
 from kedro.io import DataCatalog
-
-try:
-    # kedro 0.19.4 onwards
-    from kedro.pipeline._transcoding import TRANSCODING_SEPARATOR, _strip_transcoding
-except ImportError:  # pragma: no cover
-    # older versions
-    from kedro.pipeline.pipeline import TRANSCODING_SEPARATOR, _strip_transcoding  # type: ignore
-
 from packaging.version import parse
 
 from kedro_viz.constants import KEDRO_VERSION
+from kedro_viz.utils import TRANSCODING_SEPARATOR, _strip_transcoding
 
 try:
     # kedro 0.18.11 onwards

package/kedro_viz/integrations/kedro/hooks.py

@@ -12,15 +12,7 @@
 from kedro.io import DataCatalog
 from kedro.io.core import get_filepath_str
 
-try:
-    # kedro 0.19.4 onwards
-    from kedro.pipeline._transcoding import TRANSCODING_SEPARATOR, _strip_transcoding
-except ImportError:  # pragma: no cover
-    # older versions
-    from kedro.pipeline.pipeline import (  # type: ignore
-        TRANSCODING_SEPARATOR,
-        _strip_transcoding,
-    )
+from kedro_viz.utils import TRANSCODING_SEPARATOR, _strip_transcoding
 
 logger = logging.getLogger(__name__)
 

package/kedro_viz/models/flowchart.py

@@ -11,14 +11,6 @@
 from typing import Any, Dict, List, Optional, Set, Union, cast
 
 from kedro.pipeline.node import Node as KedroNode
-
-try:
-    # kedro 0.19.4 onwards
-    from kedro.pipeline._transcoding import TRANSCODING_SEPARATOR, _strip_transcoding
-except ImportError:  # pragma: no cover
-    # older versions
-    from kedro.pipeline.pipeline import TRANSCODING_SEPARATOR, _strip_transcoding  # type: ignore
-
 from pydantic import (
     BaseModel,
     ConfigDict,
@@ -29,6 +21,7 @@
 )
 
 from kedro_viz.models.utils import get_dataset_type
+from kedro_viz.utils import TRANSCODING_SEPARATOR, _strip_transcoding
 
 try:
     # kedro 0.18.11 onwards

package/kedro_viz/utils.py

@@ -0,0 +1,39 @@
+"""Transcoding related utility functions."""
+from typing import Tuple
+
+TRANSCODING_SEPARATOR = "@"
+
+
+def _transcode_split(element: str) -> Tuple[str, str]:
+    """Split the name by the transcoding separator.
+    If the transcoding part is missing, empty string will be put in.
+
+    Returns:
+        Node input/output name before the transcoding separator, if present.
+    Raises:
+        ValueError: Raised if more than one transcoding separator
+        is present in the name.
+    """
+    split_name = element.split(TRANSCODING_SEPARATOR)
+
+    if len(split_name) > 2:  # noqa: PLR2004
+        raise ValueError(  # pragma: no cover
+            f"Expected maximum 1 transcoding separator, found {len(split_name) - 1} "
+            f"instead: '{element}'."
+        )
+    if len(split_name) == 1:
+        split_name.append("")
+
+    return tuple(split_name)  # type: ignore
+
+
+def _strip_transcoding(element: str) -> str:
+    """Strip out the transcoding separator and anything that follows.
+
+    Returns:
+        Node input/output name before the transcoding separator, if present.
+    Raises:
+        ValueError: Raised if more than one transcoding separator
+        is present in the name.
+    """
+    return _transcode_split(element)[0]