fix!: store copies of data in MemoryStorageBackend

docs/upgrading/upgrading_to_v2.md

@@ -135,6 +135,12 @@ The storage backends now write their JSON data with camelCase keys, the same as
 
 Data written by v1 still loads, so an existing storage directory, Redis instance or database keeps working. Any tooling that reads these files directly needs to handle the camelCase keys.
 
+## `MemoryStorageBackend` stores copies
+
+`MemoryStorageBackend` stores a JSON snapshot of each dataset item and of each key-value store value with a JSON content type, serialized the same way as by `FileSystemStorageBackend`. Changing an object after you push or set it doesn't change the stored data, and pushing the same object twice stores two independent items. Reads return fresh copies too, so changing a returned item or value doesn't change the store.
+
+The memory backend now handles values the same way as the file-system backend. It raises on a value the file-system backend can't serialize, such as one with a circular reference or a tuple as a dict key. Values that JSON doesn't support, such as a `datetime`, are stored as strings, and dict keys that aren't strings become strings.
+
 ## Crawler `id` is a string
 
 The `id` argument of `BasicCrawler` and its subclasses is now a `str`, so pass a string wherever you passed an integer. `StatisticsState.stats_id` and `EventCrawlerStatusData.crawler_id` are strings too, and `crawler_id` holds the crawler's `id`. In v1, it held the Python object ID of the crawler.

src/crawlee/storage_backends/_memory/_dataset_backend.py

@@ -11,6 +11,7 @@
 from crawlee._utils.crypto import crypto_random_object_id
 from crawlee._utils.raise_if_too_many_kwargs import raise_if_too_many_kwargs
 from crawlee.storage_backends._base import DatasetBackend
+from crawlee.storage_backends._memory._utils import json_snapshot
 from crawlee.storage_backends.models import DatasetItemsListPage, DatasetMetadata
 
 if TYPE_CHECKING:
@@ -31,6 +32,9 @@ class MemoryDatasetBackend(DatasetBackend):
     The memory implementation provides fast access to data but is limited by available memory and
     does not support data sharing across different processes. It supports all dataset operations including
     sorting, filtering, and pagination, but performs them entirely in memory.
+
+    Each pushed item is stored as an independent JSON snapshot, serialized the same way as by the file-system backend,
+    and every read returns fresh copies, so mutating an item outside the dataset never changes the stored data.
     """
 
     def __init__(
@@ -45,7 +49,7 @@ def __init__(
         self._metadata = metadata
 
         self._records = list[Mapping[str, JsonSerializable]]()
-        """List to hold dataset items. Each item is a dictionary representing a record."""
+        """List to hold dataset items. Each item is a JSON snapshot of a pushed record."""
 
     @override
     async def get_metadata(self) -> DatasetMetadata:
@@ -118,12 +122,12 @@ async def purge(self) -> None:
     @override
     async def push_data(self, data: Sequence[Mapping[str, JsonSerializable]] | Mapping[str, JsonSerializable]) -> None:
         metadata = await self.get_metadata()
-        new_item_count = metadata.item_count
 
         items = data if isinstance(data, Sequence) else [data]
-        for item in items:
-            new_item_count += 1
-            await self._push_item(item)
+        # Serialize the whole batch before storing any item, so an item that fails to serialize leaves no partial push.
+        snapshots = json_snapshot(list(items))
+        self._records.extend(snapshots)
+        new_item_count = metadata.item_count + len(snapshots)
 
         await self._update_metadata(
             update_accessed_at=True,
@@ -186,7 +190,7 @@ async def get_data(
             limit=limit or (total - offset),
             total=total,
             desc=desc,
-            items=sliced_items,
+            items=json_snapshot(sliced_items),
         )
 
     @override
@@ -231,7 +235,7 @@ async def iterate_items(
         for item in sliced_items:
             if skip_empty and not item:
                 continue
-            yield item
+            yield json_snapshot(item)
 
         await self._update_metadata(update_accessed_at=True)
 
@@ -257,11 +261,3 @@ async def _update_metadata(
             self._metadata.modified_at = now
         if new_item_count is not None:
             self._metadata.item_count = new_item_count
-
-    async def _push_item(self, item: Mapping[str, JsonSerializable]) -> None:
-        """Push a single item to the dataset.
-
-        Args:
-            item: The data item to add to the dataset.
-        """
-        self._records.append(item)

src/crawlee/storage_backends/_memory/_key_value_store_backend.py

@@ -10,6 +10,7 @@
 from crawlee._utils.file import infer_mime_type
 from crawlee._utils.raise_if_too_many_kwargs import raise_if_too_many_kwargs
 from crawlee.storage_backends._base import KeyValueStoreBackend
+from crawlee.storage_backends._memory._utils import json_snapshot
 from crawlee.storage_backends.models import KeyValueStoreMetadata, KeyValueStoreRecord, KeyValueStoreRecordMetadata
 
 if TYPE_CHECKING:
@@ -26,6 +27,9 @@ class MemoryKeyValueStoreBackend(KeyValueStoreBackend):
 
     The memory implementation provides fast access to data but is limited by available memory and
     does not support data sharing across different processes.
+
+    A value with a JSON content type is stored as an independent JSON snapshot, serialized the same way as by the
+    file-system backend, and every read returns a fresh copy of it. Other values are stored as they are.
     """
 
     def __init__(
@@ -105,14 +109,23 @@ async def purge(self) -> None:
     async def get_value(self, *, key: str) -> KeyValueStoreRecord | None:
         await self._update_metadata(update_accessed_at=True)
 
-        # Return None if key doesn't exist
-        return self._records.get(key, None)
+        record = self._records.get(key, None)
+        if record is None:
+            return None
+
+        if 'application/json' in record.content_type:
+            return record.model_copy(update={'value': json_snapshot(record.value)})
+
+        return record
 
     @override
     async def set_value(self, *, key: str, value: Any, content_type: str | None = None) -> None:
         content_type = content_type or infer_mime_type(value)
         size = sys.getsizeof(value)
 
+        if 'application/json' in content_type:
+            value = json_snapshot(value)
+
         # Create and store the record
         record = KeyValueStoreRecord(
             key=key,

src/crawlee/storage_backends/_memory/_utils.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+def json_snapshot(value: Any) -> Any:
+    """Return an independent copy of `value`, round-tripped through JSON as the file-system backend serializes it.
+
+    A value the file-system backend can't serialize, such as one with a circular reference or a tuple as a dict key,
+    raises the same error here. Other objects JSON doesn't support become `str(value)`, and dict keys that aren't
+    strings are converted to strings, matching what reading the value back from the file-system backend returns.
+    """
+    return json.loads(json.dumps(value, ensure_ascii=False, default=str))

tests/unit/storage_backends/_memory/test_memory_dataset_backend.py

@@ -1,18 +1,28 @@
 from __future__ import annotations
 
 import asyncio
-from typing import TYPE_CHECKING
+import re
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
 
 import pytest
 
-from crawlee.storage_backends import MemoryStorageBackend
+from crawlee.configuration import Configuration
+from crawlee.storage_backends import FileSystemStorageBackend, MemoryStorageBackend
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
+    from pathlib import Path
 
     from crawlee.storage_backends._memory import MemoryDatasetBackend
 
 
+def circular_dict() -> dict[str, Any]:
+    value: dict[str, Any] = {}
+    value['self'] = value
+    return value
+
+
 @pytest.fixture
 async def dataset_backend() -> AsyncGenerator[MemoryDatasetBackend, None]:
     """Fixture that provides a fresh memory dataset backend for each test."""
@@ -76,3 +86,44 @@ async def test_memory_metadata_updates(dataset_backend: MemoryDatasetBackend) ->
     assert metadata.created_at == initial_created
     assert metadata.modified_at > initial_modified
     assert metadata.accessed_at > accessed_after_read
+
+
+@pytest.mark.parametrize(
+    'item',
+    [
+        pytest.param(circular_dict(), id='circular reference'),
+        pytest.param({'nested': {(1, 2): 'tuple key'}}, id='tuple key'),
+    ],
+)
+async def test_push_data_rejects_what_file_system_rejects(
+    dataset_backend: MemoryDatasetBackend,
+    tmp_path: Path,
+    item: dict[str, Any],
+) -> None:
+    """An item the file-system backend can't serialize raises the same error and stores nothing."""
+    configuration = Configuration(storage_dir=str(tmp_path))
+    fs_backend = await FileSystemStorageBackend().create_dataset_backend(configuration=configuration)
+
+    with pytest.raises((TypeError, ValueError)) as fs_error:
+        await fs_backend.push_data(item)
+
+    with pytest.raises(fs_error.type, match=re.escape(str(fs_error.value))):
+        await dataset_backend.push_data([{'valid': True}, item])
+
+    metadata = await dataset_backend.get_metadata()
+    assert metadata.item_count == 0
+    assert (await dataset_backend.get_data()).items == []
+
+
+async def test_push_data_serializes_like_file_system(dataset_backend: MemoryDatasetBackend, tmp_path: Path) -> None:
+    """Values JSON doesn't support are stored as the file-system backend reads them back."""
+    item = {'created_at': datetime(2026, 1, 1, tzinfo=UTC), 'tags': {'single'}, 1: 'int key'}
+    configuration = Configuration(storage_dir=str(tmp_path))
+    fs_backend = await FileSystemStorageBackend().create_dataset_backend(configuration=configuration)
+
+    await fs_backend.push_data(item)  # ty: ignore[invalid-argument-type]
+    await dataset_backend.push_data(item)  # ty: ignore[invalid-argument-type]
+
+    fs_items = (await fs_backend.get_data()).items
+    assert (await dataset_backend.get_data()).items == fs_items
+    assert fs_items == [{'created_at': '2026-01-01 00:00:00+00:00', 'tags': "{'single'}", '1': 'int key'}]

tests/unit/storage_backends/_memory/test_memory_kvs_backend.py

@@ -1,18 +1,28 @@
 from __future__ import annotations
 
 import asyncio
-from typing import TYPE_CHECKING
+import re
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
 
 import pytest
 
-from crawlee.storage_backends import MemoryStorageBackend
+from crawlee.configuration import Configuration
+from crawlee.storage_backends import FileSystemStorageBackend, MemoryStorageBackend
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
+    from pathlib import Path
 
     from crawlee.storage_backends._memory import MemoryKeyValueStoreBackend
 
 
+def circular_dict() -> dict[str, Any]:
+    value: dict[str, Any] = {}
+    value['self'] = value
+    return value
+
+
 @pytest.fixture
 async def kvs_backend() -> AsyncGenerator[MemoryKeyValueStoreBackend, None]:
     """Fixture that provides a fresh memory key-value store backend for each test."""
@@ -92,3 +102,44 @@ async def test_iterate_keys_skips_record_deleted_during_iteration(kvs_backend: M
         collected_keys.append(metadata.key)
 
     assert collected_keys == ['key1']
+
+
+@pytest.mark.parametrize(
+    'value',
+    [
+        pytest.param(circular_dict(), id='circular reference'),
+        pytest.param({'nested': {(1, 2): 'tuple key'}}, id='tuple key'),
+    ],
+)
+async def test_set_value_rejects_what_file_system_rejects(
+    kvs_backend: MemoryKeyValueStoreBackend,
+    tmp_path: Path,
+    value: dict[str, Any],
+) -> None:
+    """A JSON value the file-system backend can't serialize raises the same error and stores nothing."""
+    configuration = Configuration(storage_dir=str(tmp_path))
+    fs_backend = await FileSystemStorageBackend().create_kvs_backend(configuration=configuration)
+
+    with pytest.raises((TypeError, ValueError)) as fs_error:
+        await fs_backend.set_value(key='test-key', value=value)
+
+    with pytest.raises(fs_error.type, match=re.escape(str(fs_error.value))):
+        await kvs_backend.set_value(key='test-key', value=value)
+
+    assert await kvs_backend.get_value(key='test-key') is None
+
+
+async def test_set_value_serializes_like_file_system(kvs_backend: MemoryKeyValueStoreBackend, tmp_path: Path) -> None:
+    """A JSON value with types JSON doesn't support is stored as the file-system backend reads it back."""
+    value = {'created_at': datetime(2026, 1, 1, tzinfo=UTC), 1: 'int key'}
+    configuration = Configuration(storage_dir=str(tmp_path))
+    fs_backend = await FileSystemStorageBackend().create_kvs_backend(configuration=configuration)
+
+    await fs_backend.set_value(key='test-key', value=value)
+    await kvs_backend.set_value(key='test-key', value=value)
+
+    fs_record = await fs_backend.get_value(key='test-key')
+    memory_record = await kvs_backend.get_value(key='test-key')
+    assert fs_record is not None
+    assert memory_record is not None
+    assert memory_record.value == fs_record.value == {'created_at': '2026-01-01 00:00:00+00:00', '1': 'int key'}

tests/unit/storages/test_dataset.py

@@ -1183,3 +1183,39 @@ async def test_record_with_noascii_chars(dataset: Dataset) -> None:
     value = await dataset.get_data()
     assert value is not None
     assert value.items[0] == init_value
+
+
+async def test_push_data_stores_snapshot(dataset: Dataset) -> None:
+    """Mutating an item after pushing it doesn't change the stored item."""
+    item: dict[str, Any] = {'key': 'value', 'nested': {'list': [1, 2]}}
+    await dataset.push_data(item)
+
+    item['key'] = 'changed'
+    item['nested']['list'].append(3)
+
+    result = await dataset.get_data()
+    assert result.items == [{'key': 'value', 'nested': {'list': [1, 2]}}]
+
+
+async def test_push_same_object_twice_stores_independent_items(dataset: Dataset) -> None:
+    """Pushing one object twice, mutating it in between, stores two independent items."""
+    item = {'count': 1}
+    await dataset.push_data(item)
+    item['count'] = 2
+    await dataset.push_data(item)
+
+    result = await dataset.get_data()
+    assert result.items == [{'count': 1}, {'count': 2}]
+
+
+async def test_mutating_read_item_does_not_change_stored_item(dataset: Dataset) -> None:
+    """Mutating an item returned by `get_data` or `iterate_items` doesn't change the stored item."""
+    await dataset.push_data({'key': 'value'})
+
+    result = await dataset.get_data()
+    result.items[0]['key'] = 'changed by get_data'  # ty: ignore[invalid-assignment]
+    async for item in dataset.iterate_items():
+        item['key'] = 'changed by iterate_items'  # ty: ignore[invalid-assignment]
+
+    result = await dataset.get_data()
+    assert result.items == [{'key': 'value'}]

tests/unit/storages/test_key_value_store.py

@@ -1354,3 +1354,24 @@ async def insert_autosaved_value() -> None:
 
     # Should not raise `RuntimeError: dictionary changed size during iteration`.
     await asyncio.wait_for(asyncio.gather(clear_task, insert_task), timeout=5)
+
+
+async def test_set_value_stores_snapshot(kvs: KeyValueStore) -> None:
+    """Mutating a value after setting it doesn't change the stored value."""
+    value: dict[str, Any] = {'key': 'value', 'nested': {'list': [1, 2]}}
+    await kvs.set_value('test-key', value)
+
+    value['key'] = 'changed'
+    value['nested']['list'].append(3)
+
+    assert await kvs.get_value('test-key') == {'key': 'value', 'nested': {'list': [1, 2]}}
+
+
+async def test_mutating_read_value_does_not_change_stored_value(kvs: KeyValueStore) -> None:
+    """Mutating a value returned by `get_value` doesn't change the stored value."""
+    await kvs.set_value('test-key', {'key': 'value'})
+
+    value = await kvs.get_value('test-key')
+    value['key'] = 'changed'
+
+    assert await kvs.get_value('test-key') == {'key': 'value'}