From cead746402515fa9f20af7ed98c406378f0d77c9 Mon Sep 17 00:00:00 2001 From: Ankita Katiyar <110245118+ankatiyar@users.noreply.github.com> Date: Wed, 29 May 2024 12:43:42 +0100 Subject: [PATCH] Add transcoding related utility functions to Kedro viz (#1928) * Add transcoding related utility functions to Kedro viz Signed-off-by: Ankita Katiyar * no coverage Signed-off-by: Ankita Katiyar --------- Signed-off-by: Ankita Katiyar --- package/kedro_viz/data_access/managers.py | 2 +- .../data_access/repositories/catalog.py | 9 +---- package/kedro_viz/integrations/kedro/hooks.py | 10 +---- package/kedro_viz/models/flowchart.py | 9 +---- package/kedro_viz/utils.py | 39 +++++++++++++++++++ 5 files changed, 43 insertions(+), 26 deletions(-) create mode 100644 package/kedro_viz/utils.py diff --git a/package/kedro_viz/data_access/managers.py b/package/kedro_viz/data_access/managers.py index 4e4e772e5a..98fe36e78d 100644 --- a/package/kedro_viz/data_access/managers.py +++ b/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, diff --git a/package/kedro_viz/data_access/repositories/catalog.py b/package/kedro_viz/data_access/repositories/catalog.py index bfbd61a437..663e2d2230 100644 --- a/package/kedro_viz/data_access/repositories/catalog.py +++ b/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 diff --git a/package/kedro_viz/integrations/kedro/hooks.py b/package/kedro_viz/integrations/kedro/hooks.py index 32f2ffa594..f062c04ea8 100644 --- a/package/kedro_viz/integrations/kedro/hooks.py +++ b/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__) diff --git a/package/kedro_viz/models/flowchart.py b/package/kedro_viz/models/flowchart.py index b4d43d5cbc..d82dcc9bec 100644 --- a/package/kedro_viz/models/flowchart.py +++ b/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 diff --git a/package/kedro_viz/utils.py b/package/kedro_viz/utils.py new file mode 100644 index 0000000000..2d919a0c82 --- /dev/null +++ b/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]