feat: add send and ack mutations Cloud Spanner Queues

[finn-the-coder]

include unit and system tests

Fixes #17727

[gemini-code-assist]

Code Review

This pull request introduces send and ack methods to both the async and sync Batch classes to support Cloud Spanner queues, along with corresponding unit and system tests. The feedback highlights that the unit tests incorrectly compare proto-plus message fields directly to raw Python values, which should be resolved by comparing protobuf messages using helper functions. Additionally, the system tests should be optimized to use the shared_instance fixture instead of creating and deleting new Spanner instances, and they should utilize pytest.skip() to properly report skipped tests when queues are not supported.

[parthea]

==================================== ERRORS ====================================
_________________ ERROR at setup of test_db_batch_send_and_ack _________________
file /tmpfs/src/github/google-cloud-python/packages/google-cloud-spanner/tests/system/_async/test_database_api.py, line 90
  @pytest.mark.asyncio
  async def test_db_batch_send_and_ack(not_emulator, spanner_client, database_dialect, instance_config):
      import uuid
      from google.cloud.spanner_admin_instance_v1.types import spanner_instance_admin
      from google.cloud.spanner_admin_database_v1 import DatabaseDialect
      from google.api_core.exceptions import MethodNotImplemented, GoogleAPIError

      instance_id = f"test-instance-{uuid.uuid4().hex[:8]}"
      db_name = f"test-db-{uuid.uuid4().hex[:8]}"
      queue_name = f"test_queue_{uuid.uuid4().hex[:8]}"

      config_name = instance_config.name
      request = spanner_instance_admin.CreateInstanceRequest(
          parent=spanner_client.project_name,
          instance_id=instance_id,
          instance=spanner_instance_admin.Instance(
              config=config_name,
              display_name=instance_id,
              node_count=1,
              edition=spanner_instance_admin.Instance.Edition.ENTERPRISE,
          ),
      )
      print(f"instance creation request: {request}")
      operation = await spanner_client.instance_admin_api.create_instance(request=request)
      operation.result(600)

      test_instance = spanner_client.instance(instance_id, configuration_name=config_name)

      try:
          test_database = await test_instance.database(db_name, database_dialect=database_dialect)
          operation = await test_database.create()
          operation.result(300)
          print("Database created successfully!")

          try:
              # 3. Create the Queue
              if database_dialect == DatabaseDialect.POSTGRESQL:
                  queue_ddl = f"""CREATE QUEUE {queue_name} (
                      id bigint NOT NULL,
                      "Payload" varchar NOT NULL,
                      PRIMARY KEY (id)
                  )"""
              else:
                  queue_ddl = f"""CREATE QUEUE {queue_name} (
                      Id INT64 NOT NULL,
                      Payload STRING(MAX) NOT NULL
                  ) PRIMARY KEY (Id)"""

              try:
                  operation = await test_database.update_ddl([queue_ddl])
                  await operation.result(600)
              except MethodNotImplemented as e:
                  print(f"MethodNotImplemented. Skipping test because Queues are not implemented yet: {e}")
                  return
              except GoogleAPIError as e:
                  if getattr(e, 'code', None) == 501 or (getattr(e, 'grpc_status_code', None) and e.grpc_status_code.name == 'UNIMPLEMENTED') or "UNIMPLEMENTED" in str(e):
                      print(f"Skipping test because Queues are not implemented yet: {e}")
                      return
                  raise
              print("Queue created successfully.")

              # 4. Run mutations
              print("Sending message to queue...")
              async with test_database.batch() as batch:
                  batch.send(
                      queue=queue_name,
                      key=(2,),
                      payload="Hello, Queues!",
                  )
              print("Send successful.")

              print("Acking message in queue...")
              async with test_database.batch() as batch:
                  batch.ack(
                      queue=queue_name,
                      key=(2,),
                  )
              print("Ack successful.")

          finally:
              print("Dropping database...")
              await test_database.drop()
      finally:
          print("Dropping instance...")
          await test_instance.delete()
E       fixture 'not_emulator' not found
>       available fixtures: _asyncio_loop_factory, _class_scoped_runner, _function_scoped_runner, _module_scoped_runner, _package_scoped_runner, _session_scoped_runner, cache, capfd, capfdbinary, caplog, capsys, capsysbinary, capteesys, database_dialect, database_operation_timeout, databases_to_delete, doctest_namespace, event_loop_policy, instance_config, instance_configs, instance_operation_timeout, monkeypatch, not_postgres, proto_descriptor_file, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, reset_cached_apis, shared_database, shared_instance, shared_instance_id, spanner_client, subtests, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory, unused_tcp_port, unused_tcp_port_factory, unused_udp_port, unused_udp_port_factory
>       use 'pytest --fixtures [testpath]' for help on them.

[sinhasubham]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces support for Cloud Spanner Queues by adding send and ack methods to the Batch class in both synchronous and asynchronous modules, along with corresponding unit and system tests. Feedback on these changes suggests removing an accidentally committed scratchpad file (test_spanner.py), switching the instance edition in system tests from ENTERPRISE_PLUS to ENTERPRISE to minimize testing costs, and safely retrieving the gRPC status code name using getattr in system tests to prevent potential AttributeErrors.

[gemini-code-assist]

This file appears to be a temporary scratchpad or debug script that was accidentally committed. It should be removed from the repository.

[gemini-code-assist]

Using the ENTERPRISE_PLUS edition for system tests can significantly increase costs. Since Cloud Spanner Queues are also supported on the ENTERPRISE edition, consider using ENTERPRISE instead to minimize testing expenses.

Suggested change

	edition=spanner_instance_admin.Instance.Edition.ENTERPRISE_PLUS,
	edition=spanner_instance_admin.Instance.Edition.ENTERPRISE,

[gemini-code-assist]

Using the ENTERPRISE_PLUS edition for system tests can significantly increase costs. Since Cloud Spanner Queues are also supported on the ENTERPRISE edition, consider using ENTERPRISE instead to minimize testing expenses.

Suggested change

	edition=spanner_instance_admin.Instance.Edition.ENTERPRISE_PLUS,
	edition=spanner_instance_admin.Instance.Edition.ENTERPRISE,

[gemini-code-assist]

Checking e.grpc_status_code.name directly can raise an AttributeError if grpc_status_code is not a gRPC StatusCode enum (e.g., if it is an integer or None in some environments). Using getattr to safely retrieve the status code name is more robust and avoids potential runtime errors during tests.

        except GoogleAPIError as e:
            grpc_status_name = getattr(getattr(e, "grpc_status_code", None), "name", None)
            if (
                getattr(e, "code", None) == 501
                or grpc_status_name == "UNIMPLEMENTED"
                or "UNIMPLEMENTED" in str(e)
            ):
                pytest.skip(f"Queues are not implemented yet: {e}")
            raise

[gemini-code-assist]

Checking e.grpc_status_code.name directly can raise an AttributeError if grpc_status_code is not a gRPC StatusCode enum (e.g., if it is an integer or None in some environments). Using getattr to safely retrieve the status code name is more robust and avoids potential runtime errors during tests.

Suggested change

	except GoogleAPIError as e:
	if (
	getattr(e, "code", None) == 501
	or getattr(e, "grpc_status_code", None)
	and e.grpc_status_code.name == "UNIMPLEMENTED"
	or "UNIMPLEMENTED" in str(e)
	):
	pytest.skip(f"Queues are not implemented yet: {e}")
	raise
	except GoogleAPIError as e:
	grpc_status_name = getattr(getattr(e, "grpc_status_code", None), "name", None)
	if (
	getattr(e, "code", None) == 501
	or grpc_status_name == "UNIMPLEMENTED"
	or "UNIMPLEMENTED" in str(e)
	):
	pytest.skip(f"Queues are not implemented yet: {e}")
	raise

[sinhasubham]

Is this added by mistake?