refactor!: Introduce new storage client system (#1194)

## Description

- I consolidated all commits from
#1107 into this new PR.
- The previous storage-clients implementation was completely replaced
with a redesigned clients, including:
    - new storage-client interface,
    - in-memory storage client,
    - file-system storage client,
- Apify storage client (implemented in the SDK; see
apify/apify-sdk-python#470),
    - and various adjustments elsewhere in the codebase.
- The old "memory plus persist" client has been split into separate
memory and file-system implementations.
- The `Configuration.persist_storage` and
`Configuration.persist_metadata` options were removed.
- All old collection clients have been removed, they're no longer
needed.
- Each storage client now prints warnings if you pass method arguments
it does not support.
- The creation management modules in the storage clients and storages
were removed.
- Storage client parameters (e.g. `purge_on_start`, or `token` and
`base_api_url` for the Apify client) are configured via the
`Configuration`.
- Every storage, and its corresponding client, now provides both a
`purge` method (which clears all items but preserves the storage and
metadata) and a `drop` method (which removes the entire storage,
metadata included).
- All unused types, models, and helper utilities have been removed.
- The detailed, per-storage/client changes are listed below...

### Dataset

- Properties:
    - `id`
    - `name`
    - `metadata`
- Methods:
    - `open`
    - `purge` (new method)
    - `drop`
    - `push_data`
    - `get_data`
    - `iterate_items`
    - `list_items` (new method)
    - `export_to`
- Breaking changes:
- `from_storage_object` method has been removed - Use the `open` method
with `name` or `id` instead.
    - `get_info` -> `metadata` property
    - `storage_object` -> `metadata` property
- `set_metadata` method has been removed (it wasn't propage to clients)
        - Do we want to support it (e.g. for renaming)?
- `write_to_json` -> method has been removed, use `export_to` instead
    - `write_to_csv` -> method has been removed, use `export_to` instead

```python
import asyncio

from crawlee.storage_clients import FileSystemStorageClient
from crawlee.storages import Dataset


async def main() -> None:
    dataset = await Dataset.open(storage_client=FileSystemStorageClient())
    print(f'default dataset - ID: {dataset.id}, name: {dataset.name}')

    await dataset.push_data({'name': 'John'})
    await dataset.push_data({'name': 'John', 'age': 20})
    await dataset.push_data({})

    dataset_with_name = await Dataset.open(
        name='my_dataset',
        storage_client=FileSystemStorageClient(),
    )
    print(f'named dataset - ID: {dataset_with_name.id}, name: {dataset_with_name.name}')

    await dataset_with_name.push_data([{'age': 30}, {'age': 25}])

    print('Default dataset items:')
    async for item in dataset.iterate_items(skip_empty=True):
        print(item)

    print('Named dataset items:')
    async for item in dataset_with_name.iterate_items():
        print(item)

    items = await dataset.get_data()
    print(items)

    dataset_by_id = await Dataset.open(id=dataset_with_name.id)
    print(f'dataset by ID - ID: {dataset_by_id.id}, name: {dataset_by_id.name}')


if __name__ == '__main__':
    asyncio.run(main())
```

### Key-value store

- Properties:
    - `id`
    - `name`
    - `metadata`
- Methods:
    - `open`
    - `purge` (new method)
    - `drop`
    - `get_value`
    - `set_value`
- `delete_value` (new method, Apify platform's set_value support setting
an empty value to a key, so having a separate method for deleting is
useful)
    - `iterate_keys`
    - `list_keys` (new method)
    - `get_public_url`
    - `get_auto_saved_value`
    - `persist_autosaved_values`
- Breaking changes:
- `from_storage_object` method has been removed - Use the `open` method
with `name` or `id` instead.
    - `get_info` -> `metadata` property
    - `storage_object` -> `metadata` property
- `set_metadata` method has been removed (it wasn't propage to clients)
        - Do we want to support it (e.g. for renaming)?

```python
import asyncio

import requests

from crawlee.storage_clients import FileSystemStorageClient
from crawlee.storages import KeyValueStore


async def main() -> None:
    print('Opening key-value store "my_kvs"...')
    storage_client = FileSystemStorageClient()
    kvs = await KeyValueStore.open(name='my_kvs', storage_client=storage_client)

    print('Setting value to "file.json"...')
    await kvs.set_value('file.json', {'key': 'value'})

    print('Setting value to "file.jpg"...')
    response = requests.get('https://avatars.githubusercontent.com/u/25082181')
    await kvs.set_value('file.jpg', response.content)

    print('Iterating over keys:')
    async for key in kvs.iterate_keys():
        print(f'Key: {key}')

    print('Listing keys:')
    keys = [key.key for key in await kvs.list_keys()]
    print(f'Keys: {keys}')

    for key in keys:
        print(f'Getting value of {key}...')
        value = await kvs.get_value(key)
        print(f'Value: {str(value)[:100]}')

    print('Deleting value of "file.json"...')
    await kvs.delete_value('file.json')

    kvs_default = await KeyValueStore.open(storage_client=storage_client)

    special_key = 'key with spaces/and/slashes!@#$%^&*()'
    test_value = 'Special key value'

    await kvs_default.set_value(key=special_key, value=test_value)

    record = await kvs_default.get_value(key=special_key)
    assert record is not None
    assert record == test_value

    result = await kvs_default.list_keys()
    print(f'kvs_default list keys = {result}')

    kvs_2 = await KeyValueStore.open()
    result = await kvs_2.list_keys()
    print(f'kvs_2 list keys = {result}')


if __name__ == '__main__':
    asyncio.run(main())
```

### Request queue

- Properties:
    - `id`
    - `name`
    - `metadata`
- Methods:
    - `open`
    - `purge` (new method)
    - `drop`
    - `add_request`
    - `add_requests_batched` -> `add_requests`
    - `fetch_next_request`
    - `get_request`
    - `mark_request_as_handled`
    - `reclaim_request`
    - `is_empty`
    - `is_finished`
- Breaking changes:
- `from_storage_object` method has been removed - Use the `open` method
with `name` or `id` instead.
    - `get_info` -> `metadata` property
    - `storage_object` -> `metadata` property
- `set_metadata` method has been removed (it wasn't propage to clients)
        - Do we want to support it (e.g. for renaming)?
- `get_handled_count` method had been removed - Use
`metadata.handled_request_count` instead.
- `get_total_count` method has been removed - Use
`metadata.total_request_count` instead.
- `resource_directory` from the `RequestQueueMetadata` was removed, use
`path_to...` property instead.
- `RequestQueueHead` model has been removed - Use
`RequestQueueHeadWithLocks` instead.
- Notes:
- New RQ `add_requests` contain `forefront` arg (Apify API supports it)

```python
import asyncio

from crawlee import Request
from crawlee.configuration import Configuration
from crawlee.storage_clients import FileSystemStorageClient
from crawlee.storages import RequestQueue


async def main() -> None:
    rq = await RequestQueue.open(
        name='my-queue',
        storage_client=FileSystemStorageClient(),
        configuration=Configuration(purge_on_start=True),
    )

    print(f'RequestQueue: {rq}')
    print(f'RequestQueue client: {rq._client}')

    await rq.add_requests(
        requests=[
            Request.from_url('https://example.com', use_extended_unique_key=True),
            Request.from_url('https://crawlee.dev', use_extended_unique_key=True),
            Request.from_url('https://apify.com', use_extended_unique_key=True),
        ],
    )

    print('Requests were added to the queue')

    is_empty = await rq.is_empty()
    is_finished = await rq.is_finished()

    print(f'Is empty: {is_empty}')
    print(f'Is finished: {is_finished}')

    request = await rq.fetch_next_request()
    print(f'Fetched request: {request}')

    await rq.add_request('https://facebook.com', forefront=True)

    request = await rq.fetch_next_request()
    print(f'Fetched request: {request}')

    rq_default = await RequestQueue.open(
        storage_client=FileSystemStorageClient(),
        configuration=Configuration(purge_on_start=True),
    )

    await rq_default.add_request('https://example.com/1')
    await rq_default.add_requests(
        [
            'https://example.com/priority-1',
            'https://example.com/priority-2',
            'https://example.com/priority-3',
        ]
    )
    await rq_default.add_request('https://example.com/2')


if __name__ == '__main__':
    asyncio.run(main())
```

### BaseDatasetClient

- Properties:
    - `metadata`
- Methods:
    - `open`
    - `purge`
    - `drop`
    - `push_data`
    - `get_data`
    - `iterate_items`

### BaseKeyValueStoreClient

- Properties:
    - `metadata`
- Methods:
    - `open`
    - `purge`
    - `drop`
    - `get_value`
    - `set_value`
    - `delete_value`
    - `iterate_keys`
    - `get_public_url`

### BaseRequestQueueClient

- Properties:
    - `metadata`
- Methods:
    - `open`
    - `purge`
    - `drop`
- `add_requests_batch` -> `add_batch_of_requests` (one backend method
for 2 frontend methods)
    - `get_request`
    - `fetch_next_request`
    - `mark_request_as_handled`
    - `reclaim_request`
    - `is_empty`
- Models
    - `RequestQueueHeadWithLocks` -> `RequestQueueHead`
    - `BatchRequestsOperationResponse` -> `AddRequestsResponse`
- Notes:
- Old file system (memory) version didn't persist the in-progress
requests
- Old file system (memory) version didn't persist the forefront values
(now there is a FS-specific `_sequence` field in the FS Request)
- The methods manipulating locks and listing heads are now only internal
methods of Apify RQ client.

## Issues

- Closes: #92
- Closes: #147
- Closes: #783
- Closes: #1247
- Relates: #1175
- Relates: #1191

## Testing

- The original tests were mostly removed and replaced with a new ones.
- Each storage-client implementation now has its own dedicated tests at
the client level (more targeted/edge-case coverage).
- On top of that, there are storage-level tests that use a parametrized
fixture for each storage client (`file-system` and `memory`), ensuring
every storage test runs against every client implementation.

## Checklist

- [x] CI passed

Author: vdusek

docs/deployment/code_examples/google/cloud_run_example.py

@@ -5,24 +5,23 @@
 import uvicorn
 from litestar import Litestar, get
 
-from crawlee import service_locator
 from crawlee.crawlers import PlaywrightCrawler, PlaywrightCrawlingContext
-
-# highlight-start
-# Disable writing storage data to the file system
-configuration = service_locator.get_configuration()
-configuration.persist_storage = False
-configuration.write_metadata = False
-# highlight-end
+from crawlee.storage_clients import MemoryStorageClient
 
 
 @get('/')
 async def main() -> str:
     """The crawler entry point that will be called when the HTTP endpoint is accessed."""
+    # highlight-start
+    # Disable writing storage data to the file system
+    storage_client = MemoryStorageClient()
+    # highlight-end
+
     crawler = PlaywrightCrawler(
         headless=True,
         max_requests_per_crawl=10,
         browser_type='firefox',
+        storage_client=storage_client,
     )
 
     @crawler.router.default_handler

docs/deployment/code_examples/google/google_example.py

@@ -6,22 +6,21 @@
 import functions_framework
 from flask import Request, Response
 
-from crawlee import service_locator
 from crawlee.crawlers import (
     BeautifulSoupCrawler,
     BeautifulSoupCrawlingContext,
 )
-
-# highlight-start
-# Disable writing storage data to the file system
-configuration = service_locator.get_configuration()
-configuration.persist_storage = False
-configuration.write_metadata = False
-# highlight-end
+from crawlee.storage_clients import MemoryStorageClient
 
 
 async def main() -> str:
+    # highlight-start
+    # Disable writing storage data to the file system
+    storage_client = MemoryStorageClient()
+    # highlight-end
+
     crawler = BeautifulSoupCrawler(
+        storage_client=storage_client,
         max_request_retries=1,
         request_handler_timeout=timedelta(seconds=30),
         max_requests_per_crawl=10,

docs/examples/code_examples/export_entire_dataset_to_file_csv.py

@@ -30,7 +30,7 @@ async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
     await crawler.run(['https://crawlee.dev'])
 
     # Export the entire dataset to a CSV file.
-    await crawler.export_data_csv(path='results.csv')
+    await crawler.export_data(path='results.csv')
 
 
 if __name__ == '__main__':

docs/examples/code_examples/export_entire_dataset_to_file_json.py

@@ -30,7 +30,7 @@ async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
     await crawler.run(['https://crawlee.dev'])
 
     # Export the entire dataset to a JSON file.
-    await crawler.export_data_json(path='results.json')
+    await crawler.export_data(path='results.json')
 
 
 if __name__ == '__main__':

docs/examples/code_examples/parsel_crawler.py

@@ -40,7 +40,7 @@ async def some_hook(context: BasicCrawlingContext) -> None:
     await crawler.run(['https://github.com'])
 
     # Export the entire dataset to a JSON file.
-    await crawler.export_data_json(path='results.json')
+    await crawler.export_data(path='results.json')
 
 
 if __name__ == '__main__':

docs/guides/code_examples/storage_clients/custom_storage_client_example.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from crawlee.storage_clients import StorageClient
+from crawlee.storage_clients._base import (
+    DatasetClient,
+    KeyValueStoreClient,
+    RequestQueueClient,
+)
+
+if TYPE_CHECKING:
+    from crawlee.configuration import Configuration
+
+# Implement the storage type clients with your backend logic.
+
+
+class CustomDatasetClient(DatasetClient):
+    # Implement methods like push_data, get_data, iterate_items, etc.
+    pass
+
+
+class CustomKeyValueStoreClient(KeyValueStoreClient):
+    # Implement methods like get_value, set_value, delete, etc.
+    pass
+
+
+class CustomRequestQueueClient(RequestQueueClient):
+    # Implement methods like add_request, fetch_next_request, etc.
+    pass
+
+
+# Implement the storage client factory.
+
+
+class CustomStorageClient(StorageClient):
+    async def create_dataset_client(
+        self,
+        *,
+        id: str | None = None,
+        name: str | None = None,
+        configuration: Configuration | None = None,
+    ) -> CustomDatasetClient:
+        # Create and return your custom dataset client.
+        pass
+
+    async def create_kvs_client(
+        self,
+        *,
+        id: str | None = None,
+        name: str | None = None,
+        configuration: Configuration | None = None,
+    ) -> CustomKeyValueStoreClient:
+        # Create and return your custom key-value store client.
+        pass
+
+    async def create_rq_client(
+        self,
+        *,
+        id: str | None = None,
+        name: str | None = None,
+        configuration: Configuration | None = None,
+    ) -> CustomRequestQueueClient:
+        # Create and return your custom request queue client.
+        pass

docs/guides/code_examples/storage_clients/file_system_storage_client_basic_example.py

@@ -0,0 +1,8 @@
+from crawlee.crawlers import ParselCrawler
+from crawlee.storage_clients import FileSystemStorageClient
+
+# Create a new instance of storage client.
+storage_client = FileSystemStorageClient()
+
+# And pass it to the crawler.
+crawler = ParselCrawler(storage_client=storage_client)

docs/guides/code_examples/storage_clients/file_system_storage_client_configuration_example.py

@@ -0,0 +1,18 @@
+from crawlee.configuration import Configuration
+from crawlee.crawlers import ParselCrawler
+from crawlee.storage_clients import FileSystemStorageClient
+
+# Create a new instance of storage client.
+storage_client = FileSystemStorageClient()
+
+# Create a configuration with custom settings.
+configuration = Configuration(
+    storage_dir='./my_storage',
+    purge_on_start=False,
+)
+
+# And pass them to the crawler.
+crawler = ParselCrawler(
+    storage_client=storage_client,
+    configuration=configuration,
+)

docs/guides/code_examples/storage_clients/memory_storage_client_basic_example.py

@@ -0,0 +1,8 @@
+from crawlee.crawlers import ParselCrawler
+from crawlee.storage_clients import MemoryStorageClient
+
+# Create a new instance of storage client.
+storage_client = MemoryStorageClient()
+
+# And pass it to the crawler.
+crawler = ParselCrawler(storage_client=storage_client)

docs/guides/code_examples/storage_clients/registering_storage_client_example.py

@@ -0,0 +1,29 @@
+import asyncio
+
+from crawlee import service_locator
+from crawlee.crawlers import ParselCrawler
+from crawlee.storage_clients import MemoryStorageClient
+from crawlee.storages import Dataset
+
+
+async def main() -> None:
+    # Create custom storage client, MemoryStorageClient for example.
+    storage_client = MemoryStorageClient()
+
+    # Register it globally via the service locator.
+    service_locator.set_storage_client(storage_client)
+
+    # Or pass it directly to the crawler, it will be registered globally
+    # to the service locator under the hood.
+    crawler = ParselCrawler(storage_client=storage_client)
+
+    # Or just provide it when opening a storage (e.g. dataset), it will be used
+    # for this storage only, not globally.
+    dataset = await Dataset.open(
+        name='my_dataset',
+        storage_client=storage_client,
+    )
+
+
+if __name__ == '__main__':
+    asyncio.run(main())