<commit_before># Copyright 2019 Google LLC. All Rights Reserved.
#
# 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.
"""Executor specifications for defining what to to execute."""

import copy
from typing import cast, Iterable, List, Optional, Type

from tfx import types
from tfx.dsl.components.base import base_executor
from tfx.proto.orchestration import executable_spec_pb2
from tfx.utils import import_utils
from tfx.utils import json_utils
from tfx.utils import name_utils

from google.protobuf import message


class ExecutorSpec(json_utils.Jsonable):
  """A specification for a component executor.

  An instance of ExecutorSpec describes the implementation of a component.
  """

  def encode(
      self,
      component_spec: Optional[types.ComponentSpec] = None) -> message.Message:
    """Encodes ExecutorSpec into an IR proto for compiling.

    This method will be used by DSL compiler to generate the corresponding IR.

    Args:
      component_spec: Optional. The ComponentSpec to help with the encoding.

    Returns:
      An executor spec proto.
    """
    # TODO(b/158712976, b/161286496): Serialize executor specs for different
    # platforms.
    raise NotImplementedError('{} does not support encoding into IR.'.format(
        name_utils.get_full_name(self.__class__)))

  def copy(self) -> 'ExecutorSpec':
    """Makes a copy of the ExecutorSpec.

    An abstract method to implement to make a copy of the ExecutorSpec instance.
    Deepcopy is preferred in the implementation. But if for any reason a
    deepcopy is not able to be made because of some fields are not deepcopyable,
    it is OK to make a shallow copy as long as the subfield is consider
    globally immutable.

    Returns:
      A copy of ExecutorSpec.

    """
    return cast('ExecutorSpec', copy.deepcopy(self))


class ExecutorClassSpec(ExecutorSpec):
  """A specification of executor class.

  Attributes:
    executor_class: a subclass of base_executor.BaseExecutor used to execute
      this component (required).
    extra_flags: extra flags to be set in the Python base executor.
  """

  def __init__(self, executor_class: Type[base_executor.BaseExecutor]):
    if not executor_class:
      raise ValueError('executor_class is required')
    self.executor_class = executor_class
    self.extra_flags = []
    super().__init__()

  def __reduce__(self):
    # When executing on the Beam DAG runner, the ExecutorClassSpec instance
    # is pickled using the "dill" library. To make sure that the executor code
    # itself is not pickled, we save the class path which will be reimported
    # by the worker in this custom __reduce__ function.
    #
    # See https://docs.python.org/3/library/pickle.html#object.__reduce__ for
    # more details.
    return (ExecutorClassSpec._reconstruct_from_executor_class_path,
            (self.class_path,))

  @property
  def class_path(self):
    """Fully qualified class name for the executor class.

    <executor_class_module>.<executor_class_name>

    Returns:
      Fully qualified class name for the executor class.
    """
    return name_utils.get_full_name(self.executor_class)

  @staticmethod
  def _reconstruct_from_executor_class_path(executor_class_path):
    executor_class = import_utils.import_class_by_path(executor_class_path)
    return ExecutorClassSpec(executor_class)

  def encode(
      self,
      component_spec: Optional[types.ComponentSpec] = None) -> message.Message:
    result = executable_spec_pb2.PythonClassExecutableSpec()
    result.class_path = self.class_path
    result.extra_flags.extend(self.extra_flags)
    return result

  def add_extra_flags(self, extra_flags: Iterable[str]) -> None:
    self.extra_flags.extend(extra_flags)
<commit_msg>no-op<commit_after># Copyright 2019 Google LLC. All Rights Reserved.
#
# 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.
"""Executor specifications for defining what to to execute."""

import copy
from typing import cast, Iterable, List, Optional, Type

from tfx import types
from tfx.dsl.components.base import base_executor
from tfx.proto.orchestration import executable_spec_pb2
from tfx.utils import import_utils
from tfx.utils import json_utils
from tfx.utils import name_utils

from google.protobuf import message


class ExecutorSpec(json_utils.Jsonable):
  """A specification for a component executor.

  An instance of ExecutorSpec describes the implementation of a component.
  """

  def encode(
      self,
      component_spec: Optional[types.ComponentSpec] = None) -> message.Message:
    """Encodes ExecutorSpec into an IR proto for compiling.

    This method will be used by DSL compiler to generate the corresponding IR.

    Args:
      component_spec: Optional. The ComponentSpec to help with the encoding.

    Returns:
      An executor spec proto.
    """
    # TODO(b/158712976, b/161286496): Serialize executor specs for different
    # platforms.
    raise NotImplementedError('{} does not support encoding into IR.'.format(
        name_utils.get_full_name(self.__class__)))

  def copy(self) -> 'ExecutorSpec':
    """Makes a copy of the ExecutorSpec.

    An abstract method to implement to make a copy of the ExecutorSpec instance.
    Deepcopy is preferred in the implementation. But if for any reason a
    deepcopy is not able to be made because of some fields are not deepcopyable,
    it is OK to make a shallow copy as long as the subfield is consider
    globally immutable.

    Returns:
      A copy of ExecutorSpec.

    """
    return cast('ExecutorSpec', copy.deepcopy(self))


class ExecutorClassSpec(ExecutorSpec):
  """A specification of executor class.

  Attributes:
    executor_class: a subclass of base_executor.BaseExecutor used to execute
      this component (required).
    extra_flags: extra flags to be set in the Python base executor.
  """

  def __init__(self, executor_class: Type[base_executor.BaseExecutor]):
    if not executor_class:
      raise ValueError('executor_class is required')
    self.executor_class = executor_class
    self._class_path = None
    self.extra_flags = []
    super().__init__()

  def __reduce__(self):
    # When executing on the Beam DAG runner, the ExecutorClassSpec instance
    # is pickled using the "dill" library. To make sure that the executor code
    # itself is not pickled, we save the class path which will be reimported
    # by the worker in this custom __reduce__ function.
    #
    # See https://docs.python.org/3/library/pickle.html#object.__reduce__ for
    # more details.
    return (ExecutorClassSpec._reconstruct_from_executor_class_path,
            (self.class_path,))

  def _set_class_path(self):
    if self._class_path is None:
      self._class_path = name_utils.get_full_name(self.executor_class)

  @property
  def class_path(self):
    """Fully qualified class name for the executor class.

    <executor_class_module>.<executor_class_name>

    Returns:
      Fully qualified class name for the executor class.
    """
    self._set_class_path()
    return self._class_path

  @staticmethod
  def _reconstruct_from_executor_class_path(executor_class_path):
    executor_class = import_utils.import_class_by_path(executor_class_path)
    return ExecutorClassSpec(executor_class)

  def encode(
      self,
      component_spec: Optional[types.ComponentSpec] = None) -> message.Message:
    result = executable_spec_pb2.PythonClassExecutableSpec()
    result.class_path = self.class_path
    result.extra_flags.extend(self.extra_flags)
    return result

  def add_extra_flags(self, extra_flags: Iterable[str]) -> None:
    self.extra_flags.extend(extra_flags)

{'fraction_non_alphanumeric': 0.07086431226765799, 'fraction_numerical': 0.006737918215613383, 'mean_word_length': 3.82296918767507, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 15, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4216349', 'n_tokens_mistral': 2479, 'n_tokens_neox': 2385, 'n_words': 875}