{"record":{"id":"cfaa1d50b72ae3e3","repo":"apache/beam","slug":"pipeline-should-either-be-a-s-or-s-while-s-is-given","errorCode":null,"errorMessage":"pipeline should either be a %s or %s, while %s is given","messagePattern":"pipeline should either be a (.+?) or (.+?), while (.+?) is given","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/interactive/display/pipeline_graph.py","lineNumber":87,"sourceCode":"      render_option: (str) this parameter decides how the pipeline graph is\n          rendered. See display.pipeline_graph_renderer for available options.\n    \"\"\"\n    self._lock = threading.Lock()\n    self._graph: pydot.Dot = None\n    self._pipeline_instrument = None\n    if isinstance(pipeline, beam.Pipeline):\n      self._pipeline_instrument = inst.PipelineInstrument(\n          pipeline, pipeline._options)\n      # The pre-process links user pipeline to runner pipeline through analysis\n      # but without mutating runner pipeline.\n      self._pipeline_instrument.preprocess()\n\n    if isinstance(pipeline, beam_runner_api_pb2.Pipeline):\n      self._pipeline_proto = pipeline\n    elif isinstance(pipeline, beam.Pipeline):\n      self._pipeline_proto = pipeline.to_runner_api()\n    else:\n      raise TypeError(\n          'pipeline should either be a %s or %s, while %s is given' %\n          (beam_runner_api_pb2.Pipeline, beam.Pipeline, type(pipeline)))\n\n    # A dict from PCollection ID to a list of its consuming Transform IDs\n    self._consumers: collections.defaultdict[\n        str, list[str]] = collections.defaultdict(list)\n    # A dict from PCollection ID to its producing Transform ID\n    self._producers: dict[str, str] = {}\n\n    for transform_id, transform_proto in self._top_level_transforms():\n      for pcoll_id in transform_proto.inputs.values():\n        self._consumers[pcoll_id].append(transform_id)\n      for pcoll_id in transform_proto.outputs.values():\n        self._producers[pcoll_id] = transform_id\n\n    default_vertex_attrs = default_vertex_attrs or {'shape': 'box'}\n    if 'color' not in default_vertex_attrs:\n      default_vertex_attrs['color'] = 'blue'","sourceCodeStart":69,"sourceCodeEnd":105,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/interactive/display/pipeline_graph.py#L69-L105","documentation":"PipelineGraph.__init__ accepts either a runner-API Pipeline proto or an apache_beam.Pipeline object; anything else is rejected with this TypeError. It exists to fail fast when an incompatible pipeline representation is passed to interactive graph visualization.","triggerScenarios":"Constructing PipelineGraph(pipeline=...) with e.g. a str, dict, a proto from a different Beam version, a scio/other runner's object, or None instead of a beam.Pipeline or beam_runner_api_pb2.Pipeline.","commonSituations":"Building custom interactive visualizations and passing a pickled/serialized pipeline string; passing a pipeline from a mismatched apache_beam version whose to_runner_api() output type differs; passing the result of pipeline.to_runner_api() after it was JSON-converted to a dict.","solutions":["Pass an actual apache_beam.Pipeline instance or a beam_runner_api_pb2.Pipeline proto.","If you have a serialized pipeline, deserialize it back to beam_runner_api_pb2.Pipeline (proto ParseFromString) before passing.","If you have a dict, convert it with beam_runner_api_pb2.Pipeline(**d) or via json_format.ParseDict.","Check apache_beam version consistency if the object comes from another environment."],"exampleFix":"// before\ngraph = PipelineGraph(pipeline=pipeline.to_runner_api().SerializeToString())\n// after\nproto = pipeline.to_runner_api()  # beam_runner_api_pb2.Pipeline\ngraph = PipelineGraph(pipeline=proto)","handlingStrategy":"type-guard","validationCode":"assert isinstance(pipeline, (beam.Pipeline, beam_runner_api_pb2.Pipeline)), type(pipeline)","typeGuard":"def is_valid_pipeline(p) -> bool:\n    import apache_beam as beam\n    from apache_beam.pipeline import beam_runner_api_pb2\n    return isinstance(p, (beam_runner_api_pb2.Pipeline, beam.Pipeline))","tryCatchPattern":"try:\n    graph = PipelineGraph(pipeline=pipeline)\nexcept TypeError:\n    graph = PipelineGraph(pipeline=pipeline.to_runner_api() if hasattr(pipeline, 'to_runner_api') else beam_runner_api_pb2.Pipeline())","preventionTips":["Always pass beam.Pipeline or its to_runner_api() proto","Avoid serializing/deserializing pipelines as strings or dicts before graphing","Keep apache_beam versions consistent between environments"],"tags":["python","type-error","apache-beam","interactive"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}