feat: Add support for NDU storages

Author: vdusek

src/crawlee/_types.py

@@ -189,6 +189,7 @@ class PushDataFunctionCall(PushDataKwargs):
     data: list[dict[str, Any]] | dict[str, Any]
     dataset_id: str | None
     dataset_name: str | None
+    dataset_alias: str | None
 
 
 class KeyValueStoreInterface(Protocol):
@@ -255,7 +256,7 @@ def __init__(self, *, key_value_store_getter: GetKeyValueStoreFunction) -> None:
         self._key_value_store_getter = key_value_store_getter
         self.add_requests_calls = list[AddRequestsKwargs]()
         self.push_data_calls = list[PushDataFunctionCall]()
-        self.key_value_store_changes = dict[tuple[str | None, str | None], KeyValueStoreChangeRecords]()
+        self.key_value_store_changes = dict[tuple[str | None, str | None, str | None], KeyValueStoreChangeRecords]()
 
     async def add_requests(
         self,
@@ -270,6 +271,7 @@ async def push_data(
         data: list[dict[str, Any]] | dict[str, Any],
         dataset_id: str | None = None,
         dataset_name: str | None = None,
+        dataset_alias: str | None = None,
         **kwargs: Unpack[PushDataKwargs],
     ) -> None:
         """Track a call to the `push_data` context helper."""
@@ -278,6 +280,7 @@ async def push_data(
                 data=data,
                 dataset_id=dataset_id,
                 dataset_name=dataset_name,
+                dataset_alias=dataset_alias,
                 **kwargs,
             )
         )
@@ -287,13 +290,14 @@ async def get_key_value_store(
         *,
         id: str | None = None,
         name: str | None = None,
+        alias: str | None = None,
     ) -> KeyValueStoreInterface:
-        if (id, name) not in self.key_value_store_changes:
-            self.key_value_store_changes[id, name] = KeyValueStoreChangeRecords(
-                await self._key_value_store_getter(id=id, name=name)
+        if (id, name, alias) not in self.key_value_store_changes:
+            self.key_value_store_changes[id, name, alias] = KeyValueStoreChangeRecords(
+                await self._key_value_store_getter(id=id, name=name, alias=alias)
             )
 
-        return self.key_value_store_changes[id, name]
+        return self.key_value_store_changes[id, name, alias]
 
 
 @docs_group('Functions')
@@ -424,12 +428,14 @@ def __call__(
         *,
         id: str | None = None,
         name: str | None = None,
+        alias: str | None = None,
     ) -> Coroutine[None, None, KeyValueStore]:
         """Call dunder method.
 
         Args:
             id: The ID of the `KeyValueStore` to get.
-            name: The name of the `KeyValueStore` to get.
+            name: The name of the `KeyValueStore` to get (global scope).
+            alias: The alias of the `KeyValueStore` to get (run scope, unnamed).
         """
 
 
@@ -444,12 +450,14 @@ def __call__(
         *,
         id: str | None = None,
         name: str | None = None,
+        alias: str | None = None,
     ) -> Coroutine[None, None, KeyValueStoreInterface]:
         """Call dunder method.
 
         Args:
             id: The ID of the `KeyValueStore` to get.
-            name: The name of the `KeyValueStore` to get.
+            name: The name of the `KeyValueStore` to get (global scope).
+            alias: The alias of the `KeyValueStore` to get (run scope, unnamed).
         """
 
 
@@ -466,14 +474,16 @@ def __call__(
         data: list[dict[str, Any]] | dict[str, Any],
         dataset_id: str | None = None,
         dataset_name: str | None = None,
+        dataset_alias: str | None = None,
         **kwargs: Unpack[PushDataKwargs],
     ) -> Coroutine[None, None, None]:
         """Call dunder method.
 
         Args:
             data: The data to push to the `Dataset`.
             dataset_id: The ID of the `Dataset` to push the data to.
-            dataset_name: The name of the `Dataset` to push the data to.
+            dataset_name: The name of the `Dataset` to push the data to (global scope).
+            dataset_alias: The alias of the `Dataset` to push the data to (run scope, unnamed).
             **kwargs: Additional keyword arguments.
         """
 

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -557,18 +557,20 @@ async def get_dataset(
         *,
         id: str | None = None,
         name: str | None = None,
+        alias: str | None = None,
     ) -> Dataset:
         """Return the `Dataset` with the given ID or name. If none is provided, return the default one."""
-        return await Dataset.open(id=id, name=name)
+        return await Dataset.open(id=id, name=name, alias=alias)
 
     async def get_key_value_store(
         self,
         *,
         id: str | None = None,
         name: str | None = None,
+        alias: str | None = None,
     ) -> KeyValueStore:
         """Return the `KeyValueStore` with the given ID or name. If none is provided, return the default KVS."""
-        return await KeyValueStore.open(id=id, name=name)
+        return await KeyValueStore.open(id=id, name=name, alias=alias)
 
     def error_handler(
         self, handler: ErrorHandler[TCrawlingContext | BasicCrawlingContext]
@@ -772,6 +774,7 @@ async def get_data(
         self,
         dataset_id: str | None = None,
         dataset_name: str | None = None,
+        dataset_alias: str | None = None,
         **kwargs: Unpack[GetDataKwargs],
     ) -> DatasetItemsListPage:
         """Retrieve data from a `Dataset`.
@@ -781,20 +784,22 @@ async def get_data(
 
         Args:
             dataset_id: The ID of the `Dataset`.
-            dataset_name: The name of the `Dataset`.
+            dataset_name: The name of the `Dataset` (global scope).
+            dataset_alias: The alias of the `Dataset` (run scope, unnamed).
             kwargs: Keyword arguments to be passed to the `Dataset.get_data()` method.
 
         Returns:
             The retrieved data.
         """
-        dataset = await Dataset.open(id=dataset_id, name=dataset_name)
+        dataset = await Dataset.open(id=dataset_id, name=dataset_name, alias=dataset_alias)
         return await dataset.get_data(**kwargs)
 
     async def export_data(
         self,
         path: str | Path,
         dataset_id: str | None = None,
         dataset_name: str | None = None,
+        dataset_alias: str | None = None,
     ) -> None:
         """Export all items from a Dataset to a JSON or CSV file.
 
@@ -805,9 +810,10 @@ async def export_data(
         Args:
             path: The destination file path. Must end with '.json' or '.csv'.
             dataset_id: The ID of the Dataset to export from. If None, uses `name` parameter instead.
-            dataset_name: The name of the Dataset to export from. If None, uses `id` parameter instead.
+            dataset_name: The name of the Dataset to export from (global scope). If None, uses `id` parameter instead.
+            dataset_alias: The alias of the Dataset to export from (run scope, unnamed).
         """
-        dataset = await self.get_dataset(id=dataset_id, name=dataset_name)
+        dataset = await self.get_dataset(id=dataset_id, name=dataset_name, alias=dataset_alias)
 
         path = path if isinstance(path, Path) else Path(path)
         dst = path.open('w', newline='')
@@ -824,6 +830,7 @@ async def _push_data(
         data: list[dict[str, Any]] | dict[str, Any],
         dataset_id: str | None = None,
         dataset_name: str | None = None,
+        dataset_alias: str | None = None,
         **kwargs: Unpack[PushDataKwargs],
     ) -> None:
         """Push data to a `Dataset`.
@@ -834,10 +841,11 @@ async def _push_data(
         Args:
             data: The data to push to the `Dataset`.
             dataset_id: The ID of the `Dataset`.
-            dataset_name: The name of the `Dataset`.
+            dataset_name: The name of the `Dataset` (global scope).
+            dataset_alias: The alias of the `Dataset` (run scope, unnamed).
             kwargs: Keyword arguments to be passed to the `Dataset.push_data()` method.
         """
-        dataset = await self.get_dataset(id=dataset_id, name=dataset_name)
+        dataset = await self.get_dataset(id=dataset_id, name=dataset_name, alias=dataset_alias)
         await dataset.push_data(data, **kwargs)
 
     def _should_retry_request(self, context: BasicCrawlingContext, error: Exception) -> bool:
@@ -1226,8 +1234,8 @@ async def _commit_key_value_store_changes(
         result: RequestHandlerRunResult, get_kvs: GetKeyValueStoreFromRequestHandlerFunction
     ) -> None:
         """Store key value store changes recorded in result."""
-        for (id, name), changes in result.key_value_store_changes.items():
-            store = await get_kvs(id=id, name=name)
+        for (id, name, alias), changes in result.key_value_store_changes.items():
+            store = await get_kvs(id=id, name=name, alias=alias)
             for key, value in changes.updates.items():
                 await store.set_value(key, value.content, value.content_type)
 

src/crawlee/storage_clients/_base/_storage_client.py

@@ -34,6 +34,7 @@ async def create_dataset_client(
         *,
         id: str | None = None,
         name: str | None = None,
+        alias: str | None = None,
         configuration: Configuration | None = None,
     ) -> DatasetClient:
         """Create a dataset client."""
@@ -44,6 +45,7 @@ async def create_kvs_client(
         *,
         id: str | None = None,
         name: str | None = None,
+        alias: str | None = None,
         configuration: Configuration | None = None,
     ) -> KeyValueStoreClient:
         """Create a key-value store client."""
@@ -54,6 +56,7 @@ async def create_rq_client(
         *,
         id: str | None = None,
         name: str | None = None,
+        alias: str | None = None,
         configuration: Configuration | None = None,
     ) -> RequestQueueClient:
         """Create a request queue client."""

src/crawlee/storage_clients/_file_system/_dataset_client.py

@@ -58,6 +58,7 @@ def __init__(
         metadata: DatasetMetadata,
         storage_dir: Path,
         lock: asyncio.Lock,
+        directory_name: str | None = None,
     ) -> None:
         """Initialize a new instance.
 
@@ -68,6 +69,9 @@ def __init__(
         self._storage_dir = storage_dir
         """The base directory where the storage data are being persisted."""
 
+        self._directory_name = directory_name
+        """The directory name to use for this dataset. If None, uses metadata.name or default."""
+
         self._lock = lock
         """A lock to ensure that only one operation is performed at a time."""
 
@@ -78,6 +82,10 @@ async def get_metadata(self) -> DatasetMetadata:
     @property
     def path_to_dataset(self) -> Path:
         """The full path to the dataset directory."""
+        # Use the explicit directory name if provided, otherwise fall back to metadata.name or default
+        if self._directory_name is not None:
+            return self._storage_dir / self._STORAGE_SUBDIR / self._directory_name
+
         if self._metadata.name is None:
             return self._storage_dir / self._STORAGE_SUBDIR / self._STORAGE_SUBSUBDIR_DEFAULT
 
@@ -94,6 +102,7 @@ async def open(
         *,
         id: str | None,
         name: str | None,
+        alias: str | None = None,
         configuration: Configuration,
     ) -> FileSystemDatasetClient:
         """Open or create a file system dataset client.
@@ -104,15 +113,21 @@ async def open(
 
         Args:
             id: The ID of the dataset to open. If provided, searches for existing dataset by ID.
-            name: The name of the dataset to open. If not provided, uses the default dataset.
+            name: The name of the dataset for named storages. Mutually exclusive with alias.
+            alias: The alias of the dataset for unnamed storages. Mutually exclusive with name.
             configuration: The configuration object containing storage directory settings.
 
         Returns:
             An instance for the opened or created storage client.
 
         Raises:
-            ValueError: If a dataset with the specified ID is not found, or if metadata is invalid.
+            ValueError: If a dataset with the specified ID is not found, if metadata is invalid,
+                or if both name and alias are provided.
         """
+        # Validate parameters - exactly one of name or alias should be provided (or neither for default)
+        if name is not None and alias is not None:
+            raise ValueError('Cannot specify both name and alias parameters')
+
         storage_dir = Path(configuration.storage_dir)
         dataset_base_path = storage_dir / cls._STORAGE_SUBDIR
 
@@ -140,6 +155,7 @@ async def open(
                                 metadata=metadata,
                                 storage_dir=storage_dir,
                                 lock=asyncio.Lock(),
+                                directory_name=dataset_dir.name,  # Use the actual directory name
                             )
                             await client._update_metadata(update_accessed_at=True)
                             found = True
@@ -152,10 +168,27 @@ async def open(
             if not found:
                 raise ValueError(f'Dataset with ID "{id}" not found')
 
-        # Get a new instance by name.
+        # Get a new instance by name or alias.
         else:
+            # Determine the directory name and metadata name based on whether this is a named or alias storage
+            if alias is not None:
+                # For alias storages, use the alias as directory name and set metadata.name to None
+                # Special case: alias='default' should use the same directory as default storage
+                directory_name = None if alias == 'default' else alias
+                actual_name = None
+            elif name is not None:
+                # For named storages, use the name as both directory name and metadata.name
+                directory_name = name
+                actual_name = name
+            else:
+                # For default storage (no name or alias), use None for both - same as alias='default'
+                directory_name = None
+                actual_name = None
+
             dataset_path = (
-                dataset_base_path / cls._STORAGE_SUBSUBDIR_DEFAULT if name is None else dataset_base_path / name
+                dataset_base_path / cls._STORAGE_SUBSUBDIR_DEFAULT
+                if directory_name is None
+                else dataset_base_path / directory_name
             )
             metadata_path = dataset_path / METADATA_FILENAME
 
@@ -168,13 +201,17 @@ async def open(
                     await asyncio.to_thread(file.close)
                 try:
                     metadata = DatasetMetadata(**file_content)
+                    # For aliases, ensure the metadata.name is None
+                    if alias is not None:
+                        metadata = metadata.model_copy(update={'name': None})
                 except ValidationError as exc:
                     raise ValueError(f'Invalid metadata file for dataset "{name}"') from exc
 
                 client = cls(
                     metadata=metadata,
                     storage_dir=storage_dir,
                     lock=asyncio.Lock(),
+                    directory_name=directory_name,
                 )
 
                 await client._update_metadata(update_accessed_at=True)
@@ -184,7 +221,7 @@ async def open(
                 now = datetime.now(timezone.utc)
                 metadata = DatasetMetadata(
                     id=crypto_random_object_id(),
-                    name=name,
+                    name=actual_name,  # Use actual_name which will be None for aliases
                     created_at=now,
                     accessed_at=now,
                     modified_at=now,
@@ -194,6 +231,7 @@ async def open(
                     metadata=metadata,
                     storage_dir=storage_dir,
                     lock=asyncio.Lock(),
+                    directory_name=directory_name,
                 )
                 await client._update_metadata()