Skip to content

Commit

Permalink
fix(components): Remove the unused functions from function_based
Browse files Browse the repository at this point in the history
PiperOrigin-RevId: 619026632
  • Loading branch information
Googler committed Mar 27, 2024
1 parent 91ee010 commit 9a04829
Showing 1 changed file with 1 addition and 63 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
# limitations under the License.
"""Python function-based components used in KFP pipelies."""
import functools
from typing import Any, Dict, List, NamedTuple, Optional
from typing import List, NamedTuple, Optional

from google_cloud_pipeline_components import _image
from google_cloud_pipeline_components._implementation.llm import env
Expand Down Expand Up @@ -506,65 +506,3 @@ def resolve_num_microbatches(large_model_reference: str) -> int:
if 'llama' in large_model_reference.lower():
return 2
return 0


@dsl.component(base_image=_image.GCPC_IMAGE_TAG, install_kfp_package=False)
def read_file(path: str) -> str:
"""Reads the contents of the given file."""
# pylint: disable=g-import-not-at-top,import-outside-toplevel,redefined-outer-name,reimported
import re
# pylint: enable=g-import-not-at-top,import-outside-toplevel,redefined-outer-name,reimported

path = re.sub('^gs://', '/gcs/', path)
with open(path, 'r') as f:
return f.read()


@dsl.component(base_image=_image.GCPC_IMAGE_TAG, install_kfp_package=False)
def get_usage_metric(metadata: Dict[str, Any], key: str) -> bool: # pytype: disable=unsupported-operands
"""Extracts a single usage metric from metadata."""
return metadata[key]


@dsl.component(base_image=_image.GCPC_IMAGE_TAG, install_kfp_package=False)
def dump_dict(value: Dict[Any, Any]) -> str:
"""Dumps the given dict to a JSON string."""
# pylint: disable=g-import-not-at-top,import-outside-toplevel,redefined-outer-name,reimported
import json
# pylint: enable=g-import-not-at-top,import-outside-toplevel,redefined-outer-name,reimported

return json.dumps(value).replace('"', '\\"')


@dsl.component(base_image=_image.GCPC_IMAGE_TAG, install_kfp_package=False)
def dump_list(value: List[Any]) -> str:
"""Dumps the given dict to a JSON string."""
# pylint: disable=g-import-not-at-top,import-outside-toplevel,redefined-outer-name,reimported
import json
# pylint: enable=g-import-not-at-top,import-outside-toplevel,redefined-outer-name,reimported

return json.dumps(value).replace('"', '\\"')


@dsl.component(base_image=_image.GCPC_IMAGE_TAG, install_kfp_package=False)
def identity(
x: str,
) -> str:
return x


@dsl.component(base_image=_image.GCPC_IMAGE_TAG, install_kfp_package=False)
def get_uri(artifact: dsl.Input[dsl.Artifact], is_dir: bool = False) -> str: # pytype: disable=unsupported-operands
"""Extracts the URI from an artifact."""
# pylint: disable=g-import-not-at-top,import-outside-toplevel,redefined-outer-name,reimported
import os
# pylint: enable=g-import-not-at-top,import-outside-toplevel,redefined-outer-name,reimported

if is_dir:
return os.path.join(artifact.uri, '*')
return artifact.uri


@dsl.component(base_image=_image.GCPC_IMAGE_TAG, install_kfp_package=False)
def get_empty_string() -> str:
return ''

0 comments on commit 9a04829

Please sign in to comment.