Repository navigation
Expand file tree
/
Copy path_storage_client.py
More file actions
95 lines (76 loc) · 3.44 KB
/
Copy path_storage_client.py
File metadata and controls
95 lines (76 loc) · 3.44 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from crawlee._utils.docs import docs_group
if TYPE_CHECKING:
from collections.abc import Hashable
from crawlee.configuration import Configuration
from ._dataset_client import DatasetClient
from ._key_value_store_client import KeyValueStoreClient
from ._request_queue_client import RequestQueueClient
@docs_group('Storage clients')
class StorageClient(ABC):
"""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.
"""
def get_storage_client_cache_key(self, configuration: Configuration) -> Hashable: # noqa: ARG002
"""Return a cache key that can differentiate between different storages of this and other clients.
Can be based on configuration or on the client itself. By default, returns a module and name of the client
class.
"""
return f'{self.__class__.__module__}.{self.__class__.__name__}'
@abstractmethod
async def create_dataset_client(
self,
*,
id: str | None = None,
name: str | None = None,
alias: str | None = None,
configuration: Configuration | None = None,
) -> DatasetClient:
"""Create a dataset client."""
@abstractmethod
async def create_kvs_client(
self,
*,
id: str | None = None,
name: str | None = None,
alias: str | None = None,
configuration: Configuration | None = None,
) -> KeyValueStoreClient:
"""Create a key-value store client."""
@abstractmethod
async def create_rq_client(
self,
*,
id: str | None = None,
name: str | None = None,
alias: str | None = None,
configuration: Configuration | None = None,
) -> RequestQueueClient:
"""Create a 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.
"""
metadata = await client.get_metadata()
if configuration.purge_on_start and metadata.name is None:
await client.purge()