From 18070e77827b3577d2bd43f864545810b69e0f3e Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Sat, 31 Oct 2026 12:30:59 +0100 Subject: [PATCH] fix!: store copies of data in `MemoryStorageBackend` --- docs/upgrading/upgrading_to_v2.md | 6 ++ .../_memory/_dataset_backend.py | 26 ++++----- .../_memory/_key_value_store_backend.py | 17 +++++- .../storage_backends/_memory/_utils.py | 14 +++++ .../_memory/test_memory_dataset_backend.py | 55 ++++++++++++++++++- .../_memory/test_memory_kvs_backend.py | 55 ++++++++++++++++++- tests/unit/storages/test_dataset.py | 36 ++++++++++++ tests/unit/storages/test_key_value_store.py | 21 +++++++ 8 files changed, 209 insertions(+), 21 deletions(-) create mode 100644 src/crawlee/storage_backends/_memory/_utils.py diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index f7e0bd036b..44c08e3d95 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/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. diff --git a/src/crawlee/storage_backends/_memory/_dataset_backend.py b/src/crawlee/storage_backends/_memory/_dataset_backend.py index 6fac4eeec9..fcadd841f9 100644 --- a/src/crawlee/storage_backends/_memory/_dataset_backend.py +++ b/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) diff --git a/src/crawlee/storage_backends/_memory/_key_value_store_backend.py b/src/crawlee/storage_backends/_memory/_key_value_store_backend.py index 592d589a9c..a0081cdd27 100644 --- a/src/crawlee/storage_backends/_memory/_key_value_store_backend.py +++ b/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, diff --git a/src/crawlee/storage_backends/_memory/_utils.py b/src/crawlee/storage_backends/_memory/_utils.py new file mode 100644 index 0000000000..6083b90f06 --- /dev/null +++ b/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)) diff --git a/tests/unit/storage_backends/_memory/test_memory_dataset_backend.py b/tests/unit/storage_backends/_memory/test_memory_dataset_backend.py index 6368bae9e6..d5907793ee 100644 --- a/tests/unit/storage_backends/_memory/test_memory_dataset_backend.py +++ b/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'}] diff --git a/tests/unit/storage_backends/_memory/test_memory_kvs_backend.py b/tests/unit/storage_backends/_memory/test_memory_kvs_backend.py index 7c099ec820..06cd4a6054 100644 --- a/tests/unit/storage_backends/_memory/test_memory_kvs_backend.py +++ b/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'} diff --git a/tests/unit/storages/test_dataset.py b/tests/unit/storages/test_dataset.py index 401925b815..771f0a1f03 100644 --- a/tests/unit/storages/test_dataset.py +++ b/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'}] diff --git a/tests/unit/storages/test_key_value_store.py b/tests/unit/storages/test_key_value_store.py index a62b569c72..b1a3274492 100644 --- a/tests/unit/storages/test_key_value_store.py +++ b/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'}