fix!: write storage data with camelCase keys

docs/upgrading/upgrading_to_v2.md

@@ -54,3 +54,9 @@ h2 = await context.query_selector_one('h2', timedelta(seconds=5))
 await kvs.set_value('report', data, content_type='application/json')
 h2 = await context.query_selector_one('h2', timeout=timedelta(seconds=5))
 ```
+
+## Storage data uses camelCase keys
+
+The storage clients now write their JSON data with camelCase keys, the same as Crawlee JS. This covers the metadata, request and record metadata files of `FileSystemStorageClient`, the request and record metadata JSON stored by `RedisStorageClient`, and the request JSON stored by `SqlStorageClient`. For example, `item_count` in a dataset's `__metadata__.json` becomes `itemCount`. The storage metadata documents of `RedisStorageClient` keep their snake_case keys.
+
+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.

src/crawlee/storage_clients/_file_system/_dataset_client.py

@@ -436,7 +436,7 @@ async def _update_metadata(
         await asyncio.to_thread(self.path_to_metadata.parent.mkdir, parents=True, exist_ok=True)
 
         # Dump the serialized metadata to the file.
-        data = await json_dumps(self._metadata.model_dump())
+        data = await json_dumps(self._metadata.model_dump(by_alias=True))
         await atomic_write(self.path_to_metadata, data)
 
     async def _push_item(self, item: Mapping[str, JsonSerializable], item_id: int) -> None:

src/crawlee/storage_clients/_file_system/_key_value_store_client.py

@@ -324,7 +324,7 @@ async def set_value(self, *, key: str, value: Any, content_type: str | None = No
         size = len(value_bytes)
         record_metadata = KeyValueStoreRecordMetadata(key=key, content_type=content_type, size=size)
         record_metadata_filepath = record_path.with_name(f'{record_path.name}.{METADATA_FILENAME}')
-        record_metadata_content = await json_dumps(record_metadata.model_dump())
+        record_metadata_content = await json_dumps(record_metadata.model_dump(by_alias=True))
 
         async with self._lock:
             # Ensure the key-value store directory exists.
@@ -480,7 +480,7 @@ async def _update_metadata(
         await asyncio.to_thread(self.path_to_metadata.parent.mkdir, parents=True, exist_ok=True)
 
         # Dump the serialized metadata to the file.
-        data = await json_dumps(self._metadata.model_dump())
+        data = await json_dumps(self._metadata.model_dump(by_alias=True))
         await atomic_write(self.path_to_metadata, data)
 
     def _encode_key(self, key: str) -> str:

src/crawlee/storage_clients/_file_system/_request_queue_client.py

@@ -381,7 +381,7 @@ def is_in_queue(unique_key: str) -> bool:
                         state.regular_requests[request.unique_key] = sequence_number
 
                     # Save the clean request without extra fields
-                    request_data = await json_dumps(request.model_dump())
+                    request_data = await json_dumps(request.model_dump(by_alias=True))
                     await atomic_write(request_path, request_data)
 
                     # A new forefront request belongs to the very front of the queue, so it can go straight
@@ -513,7 +513,7 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
                 logger.warning(f'Request file for {request.unique_key} does not exist, cannot mark as handled.')
                 return None
 
-            request_data = await json_dumps(request.model_dump())
+            request_data = await json_dumps(request.model_dump(by_alias=True))
             await atomic_write(request_path, request_data)
 
             # Update state: remove from in-progress and pending, and add to handled. Dropping the key from
@@ -575,7 +575,7 @@ async def reclaim_request(
                 state.regular_requests[request.unique_key] = sequence_number
 
             # Save the clean request without extra fields
-            request_data = await json_dumps(request.model_dump())
+            request_data = await json_dumps(request.model_dump(by_alias=True))
             await atomic_write(request_path, request_data)
 
             # Remove from in-progress.
@@ -706,7 +706,7 @@ async def _update_metadata(
         await asyncio.to_thread(self.path_to_metadata.parent.mkdir, parents=True, exist_ok=True)
 
         # Dump the serialized metadata to the file.
-        data = await json_dumps(self._metadata.model_dump())
+        data = await json_dumps(self._metadata.model_dump(by_alias=True))
         await atomic_write(self.path_to_metadata, data)
 
     async def _refresh_cache(self) -> None:

src/crawlee/storage_clients/_redis/_key_value_store_client.py

@@ -167,7 +167,7 @@ async def set_value(self, *, key: str, value: Any, content_type: str | None = No
                 pipe.hset(
                     self._metadata_items_key,
                     key,
-                    item_metadata.model_dump_json(),
+                    item_metadata.model_dump_json(by_alias=True),
                 )
             )
             await self._update_metadata(pipe, **MetadataUpdateParams(update_accessed_at=True, update_modified_at=True))

src/crawlee/storage_clients/_redis/_request_queue_client.py

@@ -310,7 +310,7 @@ async def add_batch_of_requests(
             request = requests_by_unique_key[unique_key]
 
             new_unique_keys.append(unique_key)
-            new_request_data[unique_key] = request.model_dump_json()
+            new_request_data[unique_key] = request.model_dump_json(by_alias=True)
 
         if new_unique_keys:
             # Add new requests to the queue atomically, get back which were actually added
@@ -417,7 +417,9 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
                 elif self._dedup_strategy == 'bloom':
                     await await_redis_response(pipe.bf().add(self._handled_filter_key, request.unique_key))
 
-                await await_redis_response(pipe.hset(self._data_key, request.unique_key, request.model_dump_json()))
+                await await_redis_response(
+                    pipe.hset(self._data_key, request.unique_key, request.model_dump_json(by_alias=True))
+                )
 
                 await self._update_metadata(
                     pipe,
@@ -471,11 +473,15 @@ async def reclaim_request(
                         f'{{"client_id":"{self.client_key}","blocked_until_timestamp":{blocked_until_timestamp}}}',
                     )
                 )
-                await await_redis_response(pipe.hset(self._data_key, request.unique_key, request.model_dump_json()))
+                await await_redis_response(
+                    pipe.hset(self._data_key, request.unique_key, request.model_dump_json(by_alias=True))
+                )
                 self._pending_fetch_cache.appendleft(request)
             else:
                 await await_redis_response(pipe.rpush(self._queue_key, request.unique_key))
-                await await_redis_response(pipe.hset(self._data_key, request.unique_key, request.model_dump_json()))
+                await await_redis_response(
+                    pipe.hset(self._data_key, request.unique_key, request.model_dump_json(by_alias=True))
+                )
                 await await_redis_response(pipe.hdel(self._in_progress_key, request.unique_key))
             await self._update_metadata(
                 pipe,

src/crawlee/storage_clients/_sql/_request_queue_client.py

@@ -260,7 +260,7 @@ async def add_batch_of_requests(
                     value = {
                         'request_id': request_id,
                         'request_queue_id': self._id,
-                        'data': request.model_dump_json(),
+                        'data': request.model_dump_json(by_alias=True),
                         'is_handled': False,
                     }
                     if forefront:
@@ -306,7 +306,7 @@ async def add_batch_of_requests(
                                 'request_queue_id': self._id,
                                 'request_id': request_id,
                                 'sequence_number': state.forefront_sequence_counter,
-                                'data': request.model_dump_json(),
+                                'data': request.model_dump_json(by_alias=True),
                                 'is_handled': False,
                             }
                         )
@@ -515,7 +515,9 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
         stmt = (
             update(self._ITEM_TABLE)
             .where(self._ITEM_TABLE.request_queue_id == self._id, self._ITEM_TABLE.request_id == request_id)
-            .values(is_handled=True, time_blocked_until=None, client_key=None, data=request.model_dump_json())
+            .values(
+                is_handled=True, time_blocked_until=None, client_key=None, data=request.model_dump_json(by_alias=True)
+            )
         )
         async with self.get_session(with_simple_commit=True) as session:
             result = await session.execute(stmt)
@@ -562,7 +564,7 @@ async def reclaim_request(
                     sequence_number=new_sequence,
                     time_blocked_until=block_until,
                     client_key=self.client_key,
-                    data=request.model_dump_json(),
+                    data=request.model_dump_json(by_alias=True),
                 )
             else:
                 new_sequence = state.sequence_counter
@@ -571,7 +573,7 @@ async def reclaim_request(
                     sequence_number=new_sequence,
                     time_blocked_until=None,
                     client_key=None,
-                    data=request.model_dump_json(),
+                    data=request.model_dump_json(by_alias=True),
                 )
 
             result = await session.execute(stmt)

tests/unit/storage_clients/_file_system/test_fs_dataset_client.py

@@ -46,7 +46,7 @@ async def test_file_and_directory_creation(configuration: Configuration) -> None
         client_metadata = await client.get_metadata()
         assert metadata['id'] == client_metadata.id
         assert metadata['name'] == 'new-dataset'
-        assert metadata['item_count'] == 0
+        assert metadata['itemCount'] == 0
 
     await client.drop()
 
@@ -155,7 +155,7 @@ async def test_metadata_file_updates(dataset_client: FileSystemDatasetClient) ->
     # Verify metadata file is updated on disk
     with dataset_client.path_to_metadata.open() as f:
         metadata_json = json.load(f)
-        assert metadata_json['item_count'] == 1
+        assert metadata_json['itemCount'] == 1
 
 
 async def test_data_persistence_across_reopens() -> None:

tests/unit/storage_clients/_file_system/test_fs_kvs_client.py

@@ -9,6 +9,7 @@
 from crawlee._consts import METADATA_FILENAME
 from crawlee.configuration import Configuration
 from crawlee.storage_clients import FileSystemStorageClient
+from crawlee.storage_clients.models import KeyValueStoreMetadata, KeyValueStoreRecordMetadata
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
@@ -96,7 +97,7 @@ async def test_value_file_creation_and_content(kvs_client: FileSystemKeyValueSto
     with key_metadata_path.open() as f:
         metadata = json.load(f)
         assert metadata['key'] == test_key
-        assert metadata['content_type'] == 'text/plain; charset=utf-8'
+        assert metadata['contentType'] == 'text/plain; charset=utf-8'
         assert metadata['size'] == len(test_value.encode('utf-8'))
 
 
@@ -226,3 +227,28 @@ async def test_data_persistence_across_reopens(configuration: Configuration) ->
     assert record.value == test_value
 
     await reopened_client.drop()
+
+
+async def test_snake_case_files_from_v1_are_still_read(configuration: Configuration) -> None:
+    """Store and record metadata files with snake_case keys, as Crawlee v1 wrote them, still load."""
+    storage_client = FileSystemStorageClient()
+    client = await storage_client.create_kvs_client(name='snake-case-kvs', configuration=configuration)
+    await client.set_value(key='key', value='value')
+    kvs_id = (await client.get_metadata()).id
+
+    # Rewrite every metadata file the way v1 serialized it, without aliases.
+    for path in client.path_to_kvs.glob(f'*{METADATA_FILENAME}'):
+        model = KeyValueStoreMetadata if path == client.path_to_metadata else KeyValueStoreRecordMetadata
+        snake_case_data = model.model_validate_json(path.read_text()).model_dump(mode='json', by_alias=False)
+        path.write_text(json.dumps(snake_case_data))
+
+    reopened_client = await storage_client.create_kvs_client(id=kvs_id, configuration=configuration)
+    assert (await reopened_client.get_metadata()).id == kvs_id
+
+    record = await reopened_client.get_value(key='key')
+    assert record is not None
+    assert record.value == 'value'
+    assert record.content_type == 'text/plain; charset=utf-8'
+    assert [key.key async for key in reopened_client.iterate_keys()] == ['key']
+
+    await reopened_client.drop()

tests/unit/storage_clients/_file_system/test_fs_rq_client.py

@@ -10,6 +10,7 @@
 from crawlee import Request, service_locator
 from crawlee.configuration import Configuration
 from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient
+from crawlee.storage_clients.models import RequestQueueMetadata
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
@@ -104,6 +105,7 @@ async def test_request_file_persistence(rq_client: FileSystemRequestQueueClient)
             request_data = json.load(f)
             assert 'url' in request_data
             assert request_data['url'].startswith('https://example.com/')
+            assert 'uniqueKey' in request_data
 
 
 async def test_opening_rq_does_not_have_side_effect_on_service_locator(configuration: Configuration) -> None:
@@ -164,7 +166,7 @@ async def test_metadata_file_updates(rq_client: FileSystemRequestQueueClient) ->
     # Verify metadata file is updated on disk
     with rq_client.path_to_metadata.open() as f:
         metadata_json = json.load(f)
-        assert metadata_json['total_request_count'] == 1
+        assert metadata_json['totalRequestCount'] == 1
 
 
 async def test_data_persistence_across_reopens() -> None:
@@ -203,6 +205,34 @@ async def test_data_persistence_across_reopens() -> None:
     await reopened_client.drop()
 
 
+async def test_snake_case_files_from_v1_are_still_read() -> None:
+    """Metadata and request files with snake_case keys, as Crawlee v1 wrote them, still load."""
+    storage_client = FileSystemStorageClient()
+    client = await storage_client.create_rq_client(name='snake-case-rq')
+    request = Request.from_url('https://example.com/1', label='detail')
+    request.crawl_depth = 2
+    await client.add_batch_of_requests([request])
+    rq_id = (await client.get_metadata()).id
+
+    # Rewrite every file the way v1 serialized it, without aliases.
+    for path in client.path_to_rq.glob('*.json'):
+        data = json.loads(path.read_text())
+        model = RequestQueueMetadata if path == client.path_to_metadata else Request
+        snake_case_data = model.model_validate(data).model_dump(mode='json', by_alias=False)
+        path.write_text(json.dumps(snake_case_data))
+
+    reopened_client = await storage_client.create_rq_client(id=rq_id)
+    assert (await reopened_client.get_metadata()).total_request_count == 1
+
+    fetched = await reopened_client.fetch_next_request()
+    assert fetched is not None
+    assert fetched.unique_key == request.unique_key
+    assert fetched.label == 'detail'
+    assert fetched.crawl_depth == 2
+
+    await reopened_client.drop()
+
+
 async def test_in_progress_requests_recovered_after_crash() -> None:
     """Test that requests left in-progress at crash time are recovered as pending on reopen.
 

tests/unit/storage_clients/_redis/test_redis_kvs_client.py

@@ -77,7 +77,7 @@ async def test_value_record_creation_and_content(kvs_client: RedisKeyValueStoreC
 
     # Check record metadata
     assert metadata['key'] == test_key
-    assert metadata['content_type'] == 'text/plain; charset=utf-8'
+    assert metadata['contentType'] == 'text/plain; charset=utf-8'
     assert metadata['size'] == len(test_value.encode('utf-8'))
 
     # Verify retrieval works correctly