def debug_str()

in tfx/dsl/compiler/placeholder_utils.py [0:0]

• 102 lines of code
• 29 McCabe index (conditional complexity)

def debug_str(expression: placeholder_pb2.PlaceholderExpression) -> str:
  """Gets the debug string of a placeholder expression proto.

  Args:
    expression: A placeholder expression proto.

  Returns:
    Debug string of the placeholder expression.
  """
  if expression.HasField("value"):
    value_field_name = expression.value.WhichOneof("value")
    return f"\"{getattr(expression.value, value_field_name)}\""

  if expression.HasField("placeholder"):
    placeholder_pb = expression.placeholder
    ph_names_map = {
        placeholder_pb2.Placeholder.INPUT_ARTIFACT:
            "input",
        placeholder_pb2.Placeholder.OUTPUT_ARTIFACT:
            "output",
        placeholder_pb2.Placeholder.EXEC_PROPERTY:
            "exec_property",
        placeholder_pb2.Placeholder.RUNTIME_INFO:
            "runtime_info",
        placeholder_pb2.Placeholder.EXEC_INVOCATION:
            "execution_invocation",
        placeholder_pb2.Placeholder.ENVIRONMENT_VARIABLE:
            "environment_variable",
    }
    ph_name = ph_names_map[placeholder_pb.type]
    if placeholder_pb.key:
      return f"{ph_name}(\"{placeholder_pb.key}\")"
    else:
      return f"{ph_name}()"

  if expression.HasField("operator"):
    operator_name = expression.operator.WhichOneof("operator_type")
    operator_pb = getattr(expression.operator, operator_name)
    if operator_name == "artifact_uri_op":
      sub_expression_str = debug_str(operator_pb.expression)
      if operator_pb.split:
        return f"{sub_expression_str}.split_uri(\"{operator_pb.split}\")"
      else:
        return f"{sub_expression_str}.uri"

    if operator_name == "artifact_value_op":
      sub_expression_str = debug_str(operator_pb.expression)
      return f"{sub_expression_str}.value"

    if operator_name == "artifact_property_op":
      sub_expression_str = debug_str(operator_pb.expression)
      if operator_pb.is_custom_property:
        return f"{sub_expression_str}.custom_property(\"{operator_pb.key}\")"
      else:
        return f"{sub_expression_str}.property(\"{operator_pb.key}\")"

    if operator_name == "concat_op":
      expression_str = " + ".join(debug_str(e) for e in operator_pb.expressions)
      return f"({expression_str})"

    if operator_name == "index_op":
      sub_expression_str = debug_str(operator_pb.expression)
      return f"{sub_expression_str}[{operator_pb.index}]"

    if operator_name == "proto_op":
      sub_expression_str = debug_str(operator_pb.expression)
      field_path = "".join(operator_pb.proto_field_path)
      expression_str = f"{sub_expression_str}{field_path}"
      if operator_pb.serialization_format:
        format_str = placeholder_pb2.ProtoOperator.SerializationFormat.Name(
            operator_pb.serialization_format)
        return f"{expression_str}.serialize({format_str})"
      return expression_str

    if operator_name == "base64_encode_op":
      sub_expression_str = debug_str(operator_pb.expression)
      return f"{sub_expression_str}.b64encode()"

    if operator_name == "compare_op":
      lhs_str = debug_str(operator_pb.lhs)
      rhs_str = debug_str(operator_pb.rhs)
      if operator_pb.op == _Operation.EQUAL.value:
        op_str = "=="
      elif operator_pb.op == _Operation.LESS_THAN.value:
        op_str = "<"
      elif operator_pb.op == _Operation.GREATER_THAN.value:
        op_str = ">"
      else:
        return f"Unknown Comparison Operation {operator_pb.op}"
      return f"({lhs_str} {op_str} {rhs_str})"

    if operator_name == "unary_logical_op":
      expression_str = debug_str(operator_pb.expression)
      if operator_pb.op == _Operation.NOT.value:
        op_str = "not"
      else:
        return f"Unknown Unary Logical Operation {operator_pb.op}"
      return f"{op_str}({expression_str})"

    if operator_name == "binary_logical_op":
      lhs_str = debug_str(operator_pb.lhs)
      rhs_str = debug_str(operator_pb.rhs)
      if operator_pb.op == _Operation.AND.value:
        op_str = "and"
      elif operator_pb.op == _Operation.OR.value:
        op_str = "or"
      else:
        return f"Unknown Binary Logical Operation {operator_pb.op}"
      return f"({lhs_str} {op_str} {rhs_str})"

    if operator_name == "list_serialization_op":
      expression_str = debug_str(operator_pb.expression)
      if operator_pb.serialization_format:
        format_str = placeholder_pb2.ProtoOperator.SerializationFormat.Name(
            operator_pb.serialization_format)
        return f"{expression_str}.serialize_list({format_str})"
      return expression_str

    if operator_name == "list_concat_op":
      expression_str = ", ".join(debug_str(e) for e in operator_pb.expressions)
      return f"to_list([{expression_str}])"

    return "Unknown placeholder operator"

  return "Unknown placeholder expression"