added persistent caching for the uploaded files

pyproject.toml

@@ -25,11 +25,13 @@ packaging = ">=16.0.0"
 opentelemetry-api = ">=1.27.0"
 opentelemetry-sdk = ">=1.27.0"
 opentelemetry-exporter-otlp = ">=1.27.0"
-cachetools = "^6.2.1"
+diskcache = "^5.6.3"
 
 [tool.poetry.group.dev.dependencies]
 python-dotenv = "^1.0.1"
 black = "^24.8.0"
+ipykernel = "^7.1.0"
+jupyter = "^1.1.1"
 
 [tool.poetry.group.docs.dependencies]
 mkdocs-material = "^9.5.34"

src/rapidata/rapidata_client/config/logging_config.py

@@ -1,5 +1,6 @@
 from typing import Callable
 from pydantic import BaseModel, Field
+from rapidata.rapidata_client.config import logger
 
 # Type alias for config update handlers
 ConfigUpdateHandler = Callable[["LoggingConfig"], None]
@@ -51,8 +52,4 @@ def _notify_handlers(self) -> None:
             try:
                 handler(self)
             except Exception as e:
-                # Log the error but don't let one handler failure break others
-                print(f"Warning: Config handler failed: {e}")
-
-
-# Tracer is now handled in tracer.py with event-based updates
+                logger.warning(f"Warning: Config handler failed: {e}")

src/rapidata/rapidata_client/config/upload_config.py

@@ -1,4 +1,24 @@
+from pathlib import Path
+from typing import Callable
 from pydantic import BaseModel, Field
+from rapidata.rapidata_client.config import logger
+
+# Type alias for config update handlers
+UploadConfigUpdateHandler = Callable[["UploadConfig"], None]
+
+# Global list to store registered handlers
+_upload_config_handlers: list[UploadConfigUpdateHandler] = []
+
+
+def register_upload_config_handler(handler: UploadConfigUpdateHandler) -> None:
+    """Register a handler to be called when the upload configuration updates."""
+    _upload_config_handlers.append(handler)
+
+
+def unregister_upload_config_handler(handler: UploadConfigUpdateHandler) -> None:
+    """Unregister a previously registered handler."""
+    if handler in _upload_config_handlers:
+        _upload_config_handlers.remove(handler)
 
 
 class UploadConfig(BaseModel):
@@ -14,3 +34,22 @@ class UploadConfig(BaseModel):
     maxRetries: int = Field(default=3)
     chunkSize: int = Field(default=50)
     cacheUploads: bool = Field(default=True)
+    cacheTimeout: float = Field(default=0.1)
+    cacheLocation: Path = Field(default=Path.home() / ".rapidata" / "upload_cache")
+    cacheSizeLimit: int = Field(default=100_000_000)  # 100MB
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._notify_handlers()
+
+    def __setattr__(self, name: str, value) -> None:
+        super().__setattr__(name, value)
+        self._notify_handlers()
+
+    def _notify_handlers(self) -> None:
+        """Notify all registered handlers that the configuration has updated."""
+        for handler in _upload_config_handlers:
+            try:
+                handler(self)
+            except Exception as e:
+                logger.warning(f"Warning: UploadConfig handler failed: {e}")

src/rapidata/rapidata_client/datapoints/_asset_uploader.py

@@ -6,18 +6,61 @@
     MultiAssetInputAssetsInner,
 )
 from rapidata.api_client.models.text_asset_input import TextAssetInput
+from rapidata.rapidata_client.config.upload_config import register_upload_config_handler
+from rapidata.rapidata_client.config.upload_config import UploadConfig
 from rapidata.service.openapi_service import OpenAPIService
 from rapidata.rapidata_client.config import logger
 from rapidata.rapidata_client.config import tracer
 from rapidata.rapidata_client.config import rapidata_config
-from cachetools import LRUCache
+from diskcache import FanoutCache
+from typing import cast
 
 
 class AssetUploader:
-    _shared_upload_cache: LRUCache = LRUCache(maxsize=100_000)
+    _shared_upload_cache: FanoutCache = FanoutCache(
+        rapidata_config.upload.cacheLocation,
+        shards=rapidata_config.upload.maxWorkers,
+        timeout=rapidata_config.upload.cacheTimeout,
+        size_limit=rapidata_config.upload.cacheSizeLimit,
+    )
 
     def __init__(self, openapi_service: OpenAPIService):
         self.openapi_service = openapi_service
+        # Register the handler for config update changes
+        register_upload_config_handler(self._handle_config_update)
+
+    @classmethod
+    def _handle_config_update(cls, config: UploadConfig):
+        """
+        Handle updates to the upload config (e.g., cache location, size, workers, timeout).
+        Re-instantiates the shared cache based on the new configuration.
+        """
+        logger.debug("Updating AssetUploader shared upload cache with new config")
+        try:
+            # Dispose of the old cache (closes open resources)
+            old_cache = getattr(cls, "_shared_upload_cache", None)
+            if old_cache:
+                try:
+                    old_cache.close()
+                except Exception:
+                    pass  # ignore close errors
+
+            # Re-initialize with the updated config
+            cls._shared_upload_cache = FanoutCache(
+                config.cacheLocation,
+                shards=config.maxWorkers,
+                timeout=config.cacheTimeout,
+                size_limit=config.cacheSizeLimit,
+            )
+            logger.info(
+                "AssetUploader shared upload cache updated: location=%s, shards=%s, timeout=%s, size_limit=%s",
+                config.cacheLocation,
+                config.maxWorkers,
+                config.cacheTimeout,
+                config.cacheSizeLimit,
+            )
+        except Exception as e:
+            logger.warning(f"Failed to update AssetUploader shared upload cache: {e}")
 
     def _get_cache_key(self, asset: str) -> str:
         """Generate cache key for an asset, including environment."""
@@ -37,9 +80,10 @@ def upload_asset(self, asset: str) -> str:
             assert isinstance(asset, str), "Asset must be a string"
 
             asset_key = self._get_cache_key(asset)
-            if asset_key in self._shared_upload_cache:
+            cached_value = self._shared_upload_cache.get(asset_key)
+            if cached_value is not None:
                 logger.debug("Asset found in cache")
-                return self._shared_upload_cache[asset_key]
+                return cast(str, cached_value)  # Type hint for the linter
 
             if re.match(r"^https?://", asset):
                 response = self.openapi_service.asset_api.asset_url_post(
@@ -93,6 +137,10 @@ def get_uploaded_asset_input(
                 name=self.upload_asset(assets),
             )
 
+    def clear_cache(self):
+        self._shared_upload_cache.clear()
+        logger.info("Upload cache cleared")
+
     def __str__(self) -> str:
         return f"AssetUploader(openapi_service={self.openapi_service})"
 

src/rapidata/rapidata_client/rapidata_client.py

@@ -100,6 +100,11 @@ def reset_credentials(self):
         """Reset the credentials saved in the configuration file for the current environment."""
         self._openapi_service.reset_credentials()
 
+    def clear_all_caches(self):
+        """Clear all caches for the client."""
+        self.order._asset_uploader.clear_cache()
+        logger.info("All caches cleared")
+
     def _check_beta_features(self):
         """Enable beta features for the client."""
         with tracer.start_as_current_span("RapidataClient.check_beta_features"):