#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.
#

"""A PipelineExpansion service.
"""
# pytype: skip-file

from __future__ import absolute_import
from __future__ import print_function

import traceback

from apache_beam import pipeline as beam_pipeline
from apache_beam.portability import python_urns
from apache_beam.portability.api import beam_expansion_api_pb2
from apache_beam.portability.api import beam_expansion_api_pb2_grpc
from apache_beam.runners import pipeline_context
from apache_beam.runners.portability import portable_runner
from apache_beam.transforms import external
from apache_beam.transforms import ptransform


class ExpansionServiceServicer(
    beam_expansion_api_pb2_grpc.ExpansionServiceServicer):
  def __init__(self, options=None):
    self._options = options or beam_pipeline.PipelineOptions(
        environment_type=python_urns.EMBEDDED_PYTHON)

  def Expand(self, request, context):
    try:
      pipeline = beam_pipeline.Pipeline(options=self._options)

      def with_pipeline(component, pcoll_id=None):
        component.pipeline = pipeline
        if pcoll_id:
          component.producer, component.tag = producers[pcoll_id]
          # We need the lookup to resolve back to this id.
          context.pcollections._obj_to_id[component] = pcoll_id
        return component

      context = pipeline_context.PipelineContext(
          request.components,
          default_environment=portable_runner.PortableRunner.
          _create_environment(self._options),
          namespace=request.namespace)
      producers = {
          pcoll_id: (context.transforms.get_by_id(t_id), pcoll_tag)
          for t_id,
          t_proto in request.components.transforms.items() for pcoll_tag,
          pcoll_id in t_proto.outputs.items()
      }
      transform = with_pipeline(
          ptransform.PTransform.from_runner_api(request.transform, context))
      inputs = transform._pvaluish_from_dict({
          tag:
          with_pipeline(context.pcollections.get_by_id(pcoll_id), pcoll_id)
          for tag,
          pcoll_id in request.transform.inputs.items()
      })
      if not inputs:
        inputs = pipeline
      with external.ExternalTransform.outer_namespace(request.namespace):
        result = pipeline.apply(
            transform, inputs, request.transform.unique_name)
      expanded_transform = pipeline._root_transform().parts[-1]
      # TODO(BEAM-1833): Use named outputs internally.
      if isinstance(result, dict):
        expanded_transform.outputs = result
      pipeline_proto = pipeline.to_runner_api(context=context)
      # TODO(BEAM-1833): Use named inputs internally.
      expanded_transform_id = context.transforms.get_id(expanded_transform)
      expanded_transform_proto = pipeline_proto.components.transforms.pop(
          expanded_transform_id)
      expanded_transform_proto.inputs.clear()
      expanded_transform_proto.inputs.update(request.transform.inputs)
      for transform_id in pipeline_proto.root_transform_ids:
        del pipeline_proto.components.transforms[transform_id]
      return beam_expansion_api_pb2.ExpansionResponse(
          components=pipeline_proto.components,
          transform=expanded_transform_proto,
          requirements=pipeline_proto.requirements)

    except Exception:  # pylint: disable=broad-except
      return beam_expansion_api_pb2.ExpansionResponse(
          error=traceback.format_exc())
