feat(storage): support storage_class in AsyncAppendableObjectWriter

[chandra-siri]

Description

Adds support for specifying storage_class ('STANDARD' or 'RAPID') when writing appendable objects via AsyncAppendableObjectWriter.

Changes

• Added storage_class parameter (defaulting to None) in AsyncAppendableObjectWriter.__init__ and propagated it to _AsyncWriteObjectStream during open().
• Added storage_class parameter in _AsyncWriteObjectStream.__init__ and passed it when constructing the _storage_v2.Object resource in open().
• Added input validation in both classes ensuring storage_class must be either 'STANDARD' or 'RAPID'.
• Added unit tests in test_async_appendable_object_writer.py and test_async_write_object_stream.py covering initialization, validation, and stream opening for both 'STANDARD' and 'RAPID'.

[gemini-code-assist]

Code Review

This pull request introduces support for specifying a storage_class (specifically 'STANDARD' or 'RAPID') when asynchronously writing or appending to GCS objects. The feedback suggests importing the _SUPPORTED_STORAGE_CLASSES constant to avoid duplication, normalizing the storage_class input to uppercase for robustness, and updating the unit tests to verify this case-insensitive behavior.

[gemini-code-assist]

The constant _SUPPORTED_STORAGE_CLASSES is duplicated here and in async_write_object_stream.py. To improve maintainability and avoid potential out-of-sync issues in the future, consider importing it from google.cloud.storage.asyncio.async_write_object_stream instead of redefining it.

References

1. Remove duplicate lines of code, especially duplicate assertions in tests, to keep the codebase clean and avoid redundancy.

[gemini-code-assist]

To make the API more robust and user-friendly, consider normalizing the storage_class input to uppercase (e.g., converting 'standard' to 'STANDARD'). This prevents unexpected validation failures due to casing.

        if storage_class is not None:
            storage_class = storage_class.upper()
            if storage_class not in _SUPPORTED_STORAGE_CLASSES:
                raise ValueError(
                    f"storage_class must be either 'STANDARD' or 'RAPID', got '{storage_class}'"
                )

[gemini-code-assist]

To ensure consistency with AsyncAppendableObjectWriter and handle direct usages of _AsyncWriteObjectStream robustly, normalize the storage_class input to uppercase before validation.

        if storage_class is not None:
            storage_class = storage_class.upper()
            if storage_class not in _SUPPORTED_STORAGE_CLASSES:
                raise ValueError(
                    f"storage_class must be either 'STANDARD' or 'RAPID', got '{storage_class}'"
                )

[gemini-code-assist]

Update the unit test to verify that lowercase storage class inputs (e.g., 'standard', 'rapid') are correctly accepted and normalized to uppercase.

Suggested change

	@pytest.mark.parametrize("storage_class", ["STANDARD", "RAPID"])
	def test_init_with_storage_class(self, mock_appendable_writer, storage_class):
	writer = self._make_one(
	mock_appendable_writer["mock_client"],
	storage_class=storage_class,
	)
	assert writer.storage_class == storage_class
	@pytest.mark.parametrize(
	"storage_class, expected",
	[
	("STANDARD", "STANDARD"),
	("standard", "STANDARD"),
	("RAPID", "RAPID"),
	("rapid", "RAPID"),
	],
	)
	def test_init_with_storage_class(
	self, mock_appendable_writer, storage_class, expected
	):
	writer = self._make_one(
	mock_appendable_writer["mock_client"],
	storage_class=storage_class,
	)
	assert writer.storage_class == expected

[gemini-code-assist]

Update the unit test to verify that lowercase storage class inputs are correctly accepted and normalized to uppercase.

Suggested change

	@pytest.mark.parametrize("storage_class", ["STANDARD", "RAPID"])
	def test_init_with_storage_class(self, mock_client, storage_class):
	stream = _AsyncWriteObjectStream(
	mock_client, BUCKET, OBJECT, storage_class=storage_class
	)
	assert stream.storage_class == storage_class
	@pytest.mark.parametrize(
	"storage_class, expected",
	[
	("STANDARD", "STANDARD"),
	("standard", "STANDARD"),
	("RAPID", "RAPID"),
	("rapid", "RAPID"),
	],
	)
	def test_init_with_storage_class(self, mock_client, storage_class, expected):
	stream = _AsyncWriteObjectStream(
	mock_client, BUCKET, OBJECT, storage_class=storage_class
	)
	assert stream.storage_class == expected