diff --git a/google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py b/google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py
index be992b641..319c7964b 100644
--- a/google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py
+++ b/google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py
@@ -126,7 +126,17 @@ async def state_lookup(self) -> int:
 
     async def open(self) -> None:
         """Opens the underlying bidi-gRPC stream."""
-        raise NotImplementedError("open is not implemented yet.")
+        if self._is_stream_open:
+            raise ValueError("Underlying bidi-gRPC stream is already open")
+
+        await self.write_obj_stream.open()
+        self._is_stream_open = True
+        if self.generation is None:
+            self.generation = self.write_obj_stream.generation_number
+        self.write_handle = self.write_obj_stream.write_handle
+
+        # Update self.persisted_size
+        _ = await self.state_lookup()
 
     async def append(self, data: bytes):
         raise NotImplementedError("append is not implemented yet.")
diff --git a/tests/unit/asyncio/test_async_appendable_object_writer.py b/tests/unit/asyncio/test_async_appendable_object_writer.py
index d3a6d3830..67a074a11 100644
--- a/tests/unit/asyncio/test_async_appendable_object_writer.py
+++ b/tests/unit/asyncio/test_async_appendable_object_writer.py
@@ -112,13 +112,62 @@ async def test_state_lookup(mock_write_object_stream, mock_client):
 
 
 @pytest.mark.asyncio
-async def test_unimplemented_methods_raise_error(mock_client):
-    """Test that all currently unimplemented methods raise NotImplementedError."""
+@mock.patch(
+    "google.cloud.storage._experimental.asyncio.async_appendable_object_writer._AsyncWriteObjectStream"
+)
+async def test_open_appendable_object_writer(mock_write_object_stream, mock_client):
+    """Test the open method."""
+    # Arrange
     writer = AsyncAppendableObjectWriter(mock_client, BUCKET, OBJECT)
+    mock_stream = mock_write_object_stream.return_value
+    mock_stream.open = mock.AsyncMock()
+    mock_stream.send = mock.AsyncMock()
+    mock_stream.recv = mock.AsyncMock()
 
-    with pytest.raises(NotImplementedError):
+    mock_state_response = mock.MagicMock()
+    mock_state_response.persisted_size = 1024
+    mock_stream.recv.return_value = mock_state_response
+
+    mock_stream.generation_number = GENERATION
+    mock_stream.write_handle = WRITE_HANDLE
+
+    # Act
+    await writer.open()
+
+    # Assert
+    mock_stream.open.assert_awaited_once()
+    assert writer._is_stream_open
+    assert writer.generation == GENERATION
+    assert writer.write_handle == WRITE_HANDLE
+
+    expected_request = _storage_v2.BidiWriteObjectRequest(state_lookup=True)
+    mock_stream.send.assert_awaited_once_with(expected_request)
+    mock_stream.recv.assert_awaited_once()
+    assert writer.persisted_size == 1024
+
+
+@pytest.mark.asyncio
+@mock.patch(
+    "google.cloud.storage._experimental.asyncio.async_appendable_object_writer._AsyncWriteObjectStream"
+)
+async def test_open_when_already_open_raises_error(
+    mock_write_object_stream, mock_client
+):
+    """Test that opening an already open writer raises a ValueError."""
+    # Arrange
+    writer = AsyncAppendableObjectWriter(mock_client, BUCKET, OBJECT)
+    writer._is_stream_open = True  # Manually set to open
+
+    # Act & Assert
+    with pytest.raises(ValueError, match="Underlying bidi-gRPC stream is already open"):
         await writer.open()
 
+
+@pytest.mark.asyncio
+async def test_unimplemented_methods_raise_error(mock_client):
+    """Test that all currently unimplemented methods raise NotImplementedError."""
+    writer = AsyncAppendableObjectWriter(mock_client, BUCKET, OBJECT)
+
     with pytest.raises(NotImplementedError):
         await writer.append(b"data")
 
diff --git a/tests/unit/test_bucket.py b/tests/unit/test_bucket.py
index 8e4132edf..850e89d04 100644
--- a/tests/unit/test_bucket.py
+++ b/tests/unit/test_bucket.py
@@ -2350,7 +2350,7 @@ def test_move_blob_needs_url_encoding(self):
             timeout=30,
             retry=None,
             _target_object=new_blob,
-        )        
+        )
 
     def test_move_blob_w_user_project(self):
         source_name = "source"