{"record":{"id":"fb2925bccb8da780","repo":"apache/beam","slug":"a-sink-must-inherit-iobase-sink-iobase-nativesink-or-be-a","errorCode":null,"errorMessage":"A sink must inherit iobase.Sink, iobase.NativeSink, or be a PTransform. Received : %r","messagePattern":"A sink must inherit iobase\\.Sink, iobase\\.NativeSink, or be a PTransform\\. Received : %r","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/iobase.py","lineNumber":1127,"sourceCode":"    self.sink = sink\n\n  def display_data(self):\n    return {'sink': self.sink.__class__, 'sink_dd': self.sink}\n\n  def expand(self, pcoll):\n    # Importing locally to prevent circular dependencies.\n    from apache_beam.io.gcp.pubsub import _PubSubSink\n    if isinstance(self.sink, _PubSubSink):\n      # TODO(BEAM-27443): Remove the need for special casing here.\n      return pvalue.PDone(pcoll.pipeline)\n    elif isinstance(self.sink, Sink):\n      # A custom sink\n      return pcoll | WriteImpl(self.sink)\n    elif isinstance(self.sink, ptransform.PTransform):\n      # This allows \"composite\" sinks to be used like non-composite ones.\n      return pcoll | self.sink\n    else:\n      raise ValueError(\n          'A sink must inherit iobase.Sink, iobase.NativeSink, '\n          'or be a PTransform. Received : %r' % self.sink)\n\n  def to_runner_api_parameter(\n      self,\n      context: PipelineContext,\n  ) -> tuple[str, Any]:\n    # TODO(BEAM-27443): Remove the need for special casing here.\n    # Importing locally to prevent circular dependencies.\n    from apache_beam.io.gcp.pubsub import _PubSubSink\n    if isinstance(self.sink, _PubSubSink):\n      payload = beam_runner_api_pb2.PubSubWritePayload(\n          topic=self.sink.full_topic,\n          id_attribute=self.sink.id_label,\n          timestamp_attribute=self.sink.timestamp_attribute)\n      return (common_urns.composites.PUBSUB_WRITE.urn, payload)\n    else:\n      return super().to_runner_api_parameter(context)","sourceCodeStart":1109,"sourceCodeEnd":1145,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/iobase.py#L1109-L1145","documentation":"apache_beam.io.iobase.Write.expand validates that the sink passed to WriteToTransform is either an iobase.Sink, a NativeSink, or a PTransform. Any other object (a class instead of instance, a string, a connector config object) cannot be written to and raises ValueError. This guards the two supported write paths: custom Sink (wrapped in WriteImpl) and composite PTransform sinks.","triggerScenarios":"Calling WriteToTransform(sink).expand(pcoll) where sink is not an instance of iobase.Sink, iobase.NativeSink, or ptransform.PTransform - e.g. passing the sink class instead of an instance, or passing a connector-specific config/options object instead of the sink itself.","commonSituations":"Writing ReadAllResults with a hand-built WriteToTransform; copying code where the sink class name was passed rather than an instance; mixing up a custom IO's config object with its Sink implementation; typos in constructor calls.","solutions":["Instantiate the sink: pass MySink(...) (an instance) rather than MySink.","Ensure the sink class inherits from apache_beam.io.iobase.Sink (or NativeSink) and implements write/build_writer.","If the sink is a composite write, make it a PTransform subclass so it is accepted directly.","Prefer the connector's public write transform (e.g. WriteToText, WriteToBigQuery) instead of building WriteToTransform manually."],"exampleFix":"# before\nresult | WriteToTransform(WriteToText)  # class passed, not instance\n# after\nresult | WriteToTransform(WriteToText('/tmp/out'))  # sink instance","handlingStrategy":"type-guard","validationCode":"from apache_beam.io import iobase\nassert isinstance(sink, (iobase.Sink, iobase.NativeSink)) or isinstance(sink, beam.transforms.ptransform.PTransform)","typeGuard":"def is_valid_sink(sink) -> bool:\n    from apache_beam.io import iobase\n    from apache_beam.transforms import ptransform\n    return isinstance(sink, (iobase.Sink, iobase.NativeSink, ptransform.PTransform))","tryCatchPattern":"try:\n    expanded = WriteToTransform(sink).expand(pcoll)\nexcept ValueError as e:\n    if 'A sink must inherit' in str(e): log.error('Pass a Sink/PTransform instance, got %r', sink)","preventionTips":["Always pass sink instances, never classes","Prefer built-in write transforms over WriteToTransform","Verify custom sinks inherit iobase.Sink in unit tests"],"tags":["apache-beam","python","io","type-mismatch","pipeline-construction"],"backgroundTag":"invalid-argument-value","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"}