{"record":{"id":"39cd11804966fe0a","repo":"apache/beam","slug":"invalid-transaction-object-s-it-should-be-instance-of","errorCode":null,"errorMessage":"Invalid transaction object: %s. It should be instance of SPANNER_TRANSACTION object created by spannerio.create_transaction transform.","messagePattern":"Invalid transaction object: (.+?)\\. It should be instance of SPANNER_TRANSACTION object created by spannerio\\.create_transaction transform\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/gcp/experimental/spannerio.py","lineNumber":402,"sourceCode":"    return self._session\n\n  def _close_session(self):\n    if self._session is not None:\n      self._session.delete()\n\n  def setup(self):\n    # setting up client to connect with cloud spanner\n    spanner_client = Client(self._spanner_configuration.project)\n    instance = spanner_client.instance(self._spanner_configuration.instance)\n    self._database = instance.database(\n        self._spanner_configuration.database,\n        pool=self._spanner_configuration.pool)\n\n  def process(self, element, spanner_transaction):\n    # `spanner_transaction` should be the instance of the _SPANNER_TRANSACTION\n    # object.\n    if not isinstance(spanner_transaction, _SPANNER_TRANSACTION):\n      raise ValueError(\n          \"Invalid transaction object: %s. It should be instance \"\n          \"of SPANNER_TRANSACTION object created by \"\n          \"spannerio.create_transaction transform.\" % type(spanner_transaction))\n\n    transaction_info = spanner_transaction.transaction\n\n    # We used batch snapshot to reuse the same transaction passed through the\n    # side input\n    self._snapshot = BatchSnapshot.from_dict(self._database, transaction_info)\n\n    # getting the transaction from the snapshot's session to run read operation.\n    # with self._snapshot.session().transaction() as transaction:\n    with self._get_session().transaction() as transaction:\n      table_id = self._spanner_configuration.table\n      query_name = self._spanner_configuration.query_name or ''\n\n      if element.is_sql is True:\n        transaction_read = transaction.execute_sql","sourceCodeStart":384,"sourceCodeEnd":420,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/gcp/experimental/spannerio.py#L384-L420","documentation":"Beam's SpannerIO write transform expects the transaction side-input to be the internal _SPANNER_TRANSACTION wrapper produced by the spannerio.create_transaction transform. process() raises ValueError when the second element is anything else, because transaction_info is extracted from that wrapper's .transaction field.","triggerScenarios":"Passing a raw google.cloud.spanner.Transaction, a spanner Database object, or None as the side input to the write (or spannerio.Write / write mutations) PTransform instead of the output of create_transaction.","commonSituations":"Wiring a PCollection to the write transform without the CreateTransaction step in the pipeline; refactoring code and substituting a client-created transaction for the Beam wrapper.","solutions":["Add spannerio.create_transaction(database_id) to the pipeline and use its PCollection output as the transaction side input to the write transform.","If you have a raw Transaction, you cannot pass it directly - route writes through create_transaction so Beam manages transaction lifecycle.","Check the write pipeline ordering: write_bundles/process receives (mutation, transaction) pairs, so zip your mutations with the create_transaction output."],"exampleFix":"// before\nrows | 'write' >> spannerio.Write().with_transaction(raw_transaction)\n// after\ntransactions = beam.pvalue.AsSingleton(\n    pcoll | 'txn' >> spannerio.create_transaction(database_id))\nrows | 'write' >> spannerio.Write(database_id).with_transaction(transactions)","handlingStrategy":"type-guard","validationCode":"from apache_beam.io.gcp.experimental.spannerio import _SPANNER_TRANSACTION\nassert isinstance(txn, _SPANNER_TRANSACTION), 'use spannerio.create_transaction output'","typeGuard":"def is_spanner_transaction(obj):\n    from apache_beam.io.gcp.experimental.spannerio import _SPANNER_TRANSACTION\n    return isinstance(obj, _SPANNER_TRANSACTION)","tryCatchPattern":"try:\n    _ | write_transform\nexcept ValueError as e:\n    if 'Invalid transaction object' in str(e):\n        raise RuntimeError('Wire the CreateTransaction transform output as side input') from e\n    raise","preventionTips":["Always insert spannerio.create_transaction into the write pipeline","Never pass client-SDK Transaction objects into Beam transforms","Zip mutations with the create_transaction PCollection as singleton side input"],"tags":["google-cloud-spanner","side-input","apache-beam"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}