def _transform()

in pipelines/marketing_intelligence/marketing_intelligence_pipeline/pipeline.py [0:0]

• 8 lines of code
• 1 McCabe index (conditional complexity)

def _transform(msgs: PCollection[str], model_endpoint: str, project: str,
               location: str) -> PCollection[str]:
  model_handler = VertexAIModelHandlerJSON(
      endpoint_id=model_endpoint, project=project, location=location)
  preds: PCollection[
      PredictionResult] = msgs | "RunInference-vertexai" >> RunInference(
          model_handler)
  return preds | "Format Output" >> beam.Map(_format_output)