Add purge_if_needed method and improve some typing based on Pylance

Author: vdusek

src/crawlee/_utils/docs.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
-from typing import Callable, Literal
+from typing import Any, Callable, Literal, TypeVar
 
 GroupName = Literal['Classes', 'Abstract classes', 'Data structures', 'Event payloads', 'Errors', 'Functions']
 
+T = TypeVar('T', bound=Callable[..., Any])
 
-def docs_group(group_name: GroupName) -> Callable:  # noqa: ARG001
+
+def docs_group(group_name: GroupName) -> Callable[[T], T]:  # noqa: ARG001
     """Mark a symbol for rendering and grouping in documentation.
 
     This decorator is used solely for documentation purposes and does not modify the behavior
@@ -18,7 +20,7 @@ def docs_group(group_name: GroupName) -> Callable:  # noqa: ARG001
         The original callable without modification.
     """
 
-    def wrapper(func: Callable) -> Callable:
+    def wrapper(func: T) -> T:
         return func
 
     return wrapper

src/crawlee/_utils/file.py

@@ -128,7 +128,7 @@ def _sync_write() -> str:
 
 
 async def export_json_to_stream(
-    iterator: AsyncIterator[dict],
+    iterator: AsyncIterator[dict[str, Any]],
     dst: TextIO,
     **kwargs: Unpack[ExportDataJsonKwargs],
 ) -> None:
@@ -137,7 +137,7 @@ async def export_json_to_stream(
 
 
 async def export_csv_to_stream(
-    iterator: AsyncIterator[dict],
+    iterator: AsyncIterator[dict[str, Any]],
     dst: TextIO,
     **kwargs: Unpack[ExportDataCsvKwargs],
 ) -> None:

src/crawlee/request_loaders/_request_list.py

@@ -54,13 +54,13 @@ def __init__(
     def name(self) -> str | None:
         return self._name
 
-    @override
     @property
+    @override
     async def handled_count(self) -> int:
         return self._handled_count
 
-    @override
     @property
+    @override
     async def total_count(self) -> int:
         return self._assumed_total_count
 

src/crawlee/request_loaders/_request_manager_tandem.py

@@ -32,13 +32,13 @@ def __init__(self, request_loader: RequestLoader, request_manager: RequestManage
         self._read_only_loader = request_loader
         self._read_write_manager = request_manager
 
-    @override
     @property
+    @override
     async def handled_count(self) -> int:
         return await self._read_write_manager.handled_count
 
-    @override
     @property
+    @override
     async def total_count(self) -> int:
         return (await self._read_only_loader.total_count) + (await self._read_write_manager.total_count)
 

src/crawlee/storage_clients/_base/_dataset_client.py

@@ -112,7 +112,7 @@ async def iterate_items(
         unwind: str | None = None,
         skip_empty: bool = False,
         skip_hidden: bool = False,
-    ) -> AsyncIterator[dict]:
+    ) -> AsyncIterator[dict[str, Any]]:
         """Iterate over the dataset items with filtering options.
 
         The backend method for the `Dataset.iterate_items` call.

src/crawlee/storage_clients/_base/_storage_client.py

@@ -3,6 +3,8 @@
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING
 
+from crawlee._utils.docs import docs_group
+
 if TYPE_CHECKING:
     from crawlee.configuration import Configuration
 
@@ -11,8 +13,20 @@
     from ._request_queue_client import RequestQueueClient
 
 
+@docs_group('Abstract classes')
 class StorageClient(ABC):
-    """Base class for storage clients."""
+    """Base class for storage clients.
+
+    The `StorageClient` serves as an abstract base class that defines the interface for accessing Crawlee's
+    storage types: datasets, key-value stores, and request queues. It provides methods to open clients for
+    each of these storage types and handles common functionality.
+
+    Storage clients implementations can be provided for various backends (file system, memory, databases,
+    various cloud providers, etc.) to support different use cases from development to production environments.
+
+    Each storage client implementation is responsible for ensuring proper initialization, data persistence
+    (where applicable), and consistent access patterns across all storage types it supports.
+    """
 
     @abstractmethod
     async def open_dataset_client(
@@ -47,3 +61,21 @@ async def open_request_queue_client(
     def get_rate_limit_errors(self) -> dict[int, int]:
         """Return statistics about rate limit errors encountered by the HTTP client in storage client."""
         return {}
+
+    async def _purge_if_needed(
+        self,
+        client: DatasetClient | KeyValueStoreClient | RequestQueueClient,
+        configuration: Configuration,
+    ) -> None:
+        """Purge the client if needed.
+
+        The purge is only performed if the configuration indicates that it should be done and the client
+        is not a named storage. Named storages are considered global and will typically outlive the run,
+        so they are not purged.
+
+        Args:
+            client: The storage client to potentially purge.
+            configuration: Configuration that determines whether purging should occur.
+        """
+        if configuration.purge_on_start and client.metadata.name is None:
+            await client.purge()

src/crawlee/storage_clients/_file_system/_dataset_client.py

@@ -6,7 +6,7 @@
 from datetime import datetime, timezone
 from logging import getLogger
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from pydantic import ValidationError
 from typing_extensions import override
@@ -19,7 +19,6 @@
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
-    from typing import Any
 
     from crawlee.configuration import Configuration
 
@@ -83,8 +82,8 @@ def __init__(
         self._lock = asyncio.Lock()
         """A lock to ensure that only one operation is performed at a time."""
 
-    @override
     @property
+    @override
     def metadata(self) -> DatasetMetadata:
         return self._metadata
 
@@ -258,7 +257,7 @@ async def get_data(
         view: str | None = None,
     ) -> DatasetItemsListPage:
         # Check for unsupported arguments and log a warning if found.
-        unsupported_args = {
+        unsupported_args: dict[str, Any] = {
             'clean': clean,
             'fields': fields,
             'omit': omit,
@@ -307,7 +306,7 @@ async def get_data(
             selected_files = selected_files[:limit]
 
         # Read and parse each data file.
-        items = []
+        items = list[dict[str, Any]]()
         for file_path in selected_files:
             try:
                 file_content = await asyncio.to_thread(file_path.read_text, encoding='utf-8')
@@ -353,9 +352,9 @@ async def iterate_items(
         unwind: str | None = None,
         skip_empty: bool = False,
         skip_hidden: bool = False,
-    ) -> AsyncIterator[dict]:
+    ) -> AsyncIterator[dict[str, Any]]:
         # Check for unsupported arguments and log a warning if found.
-        unsupported_args = {
+        unsupported_args: dict[str, Any] = {
             'clean': clean,
             'fields': fields,
             'omit': omit,

src/crawlee/storage_clients/_file_system/_key_value_store_client.py

@@ -79,8 +79,8 @@ def __init__(
         self._lock = asyncio.Lock()
         """A lock to ensure that only one operation is performed at a time."""
 
-    @override
     @property
+    @override
     def metadata(self) -> KeyValueStoreMetadata:
         return self._metadata
 

src/crawlee/storage_clients/_file_system/_request_queue_client.py

@@ -99,8 +99,8 @@ def __init__(
         self._sequence_counter = 0
         """A counter to track the order of requests added to the queue."""
 
-    @override
     @property
+    @override
     def metadata(self) -> RequestQueueMetadata:
         return self._metadata
 

src/crawlee/storage_clients/_file_system/_storage_client.py

@@ -2,6 +2,7 @@
 
 from typing_extensions import override
 
+from crawlee._utils.docs import docs_group
 from crawlee.configuration import Configuration
 from crawlee.storage_clients._base import StorageClient
 
@@ -10,8 +11,20 @@
 from ._request_queue_client import FileSystemRequestQueueClient
 
 
+@docs_group('Classes')
 class FileSystemStorageClient(StorageClient):
-    """File system storage client."""
+    """File system implementation of the storage client.
+
+    This storage client provides access to datasets, key-value stores, and request queues that persist data
+    to the local file system. Each storage type is implemented with its own specific file system client
+    that stores data in a structured directory hierarchy.
+
+    Data is stored in JSON format in predictable file paths, making it easy to inspect and manipulate
+    the stored data outside of the Crawlee application if needed.
+
+    All data persists between program runs but is limited to access from the local machine
+    where the files are stored.
+    """
 
     @override
     async def open_dataset_client(
@@ -23,10 +36,7 @@ async def open_dataset_client(
     ) -> FileSystemDatasetClient:
         configuration = configuration or Configuration.get_global_configuration()
         client = await FileSystemDatasetClient.open(id=id, name=name, configuration=configuration)
-
-        if configuration.purge_on_start and client.metadata.name is None:
-            await client.purge()
-
+        await self._purge_if_needed(client, configuration)
         return client
 
     @override
@@ -39,10 +49,7 @@ async def open_key_value_store_client(
     ) -> FileSystemKeyValueStoreClient:
         configuration = configuration or Configuration.get_global_configuration()
         client = await FileSystemKeyValueStoreClient.open(id=id, name=name, configuration=configuration)
-
-        if configuration.purge_on_start and client.metadata.name is None:
-            await client.purge()
-
+        await self._purge_if_needed(client, configuration)
         return client
 
     @override
@@ -55,8 +62,5 @@ async def open_request_queue_client(
     ) -> FileSystemRequestQueueClient:
         configuration = configuration or Configuration.get_global_configuration()
         client = await FileSystemRequestQueueClient.open(id=id, name=name, configuration=configuration)
-
-        if configuration.purge_on_start and client.metadata.name is None:
-            await client.purge()
-
+        await self._purge_if_needed(client, configuration)
         return client