Storage Manager

manager

Storage manager for coordinating multiple storage backends.

The storage manager provides a unified interface for working with different storage backends and supports features like caching, fallback, and replication.

Classes

StorageManager

StorageManager(primary: StorageBackend, secondary: StorageBackend | None = None, use_cache: bool = True, cache_size_mb: float | None = 100.0)

Manager for coordinating multiple storage backends.

Supports: - Primary/secondary backend configuration - In-memory caching for fast access - Automatic fallback to secondary storage - Result migration between backends

Initialize storage manager.

Parameters:

Name	Type	Description	Default
primary	StorageBackend	Primary storage backend	required
secondary	StorageBackend | None	Optional secondary/backup storage backend	None
use_cache	bool	Enable in-memory caching	True
cache_size_mb	float | None	Maximum cache size in MB	100.0

Source code in optiscope/storage/manager.py

def __init__(
    self,
    primary: StorageBackend,
    secondary: StorageBackend | None = None,
    use_cache: bool = True,
    cache_size_mb: float | None = 100.0,
) -> None:
    """
    Initialize storage manager.

    Args:
        primary: Primary storage backend
        secondary: Optional secondary/backup storage backend
        use_cache: Enable in-memory caching
        cache_size_mb: Maximum cache size in MB
    """
    self.primary = primary
    self.secondary = secondary
    self.use_cache = use_cache

    # Initialize cache if enabled
    if use_cache:
        self.cache = MemoryStorage(max_memory_mb=cache_size_mb)
    else:
        self.cache = None

Functions

save_result

save_result(key: str, result: OptimizationResult, metadata: dict[str, Any] | None = None, replicate: bool = True) -> None

Save result to storage.

Parameters:

Name	Type	Description	Default
key	str	Unique identifier for the result	required
result	OptimizationResult	OptimizationResult to save	required
metadata	dict[str, Any] | None	Optional metadata	None
replicate	bool	Also save to secondary backend if available	True

Source code in optiscope/storage/manager.py

def save_result(
    self,
    key: str,
    result: OptimizationResult,
    metadata: dict[str, Any] | None = None,
    replicate: bool = True,
) -> None:
    """
    Save result to storage.

    Args:
        key: Unique identifier for the result
        result: OptimizationResult to save
        metadata: Optional metadata
        replicate: Also save to secondary backend if available
    """
    # Save to primary
    try:
        self.primary.save_result(key, result, metadata)
    except Exception as e:
        logger.error(f"Failed to save to primary storage: {e}")

        # Try secondary as fallback
        if self.secondary:
            logger.info("Attempting save to secondary storage")
            self.secondary.save_result(key, result, metadata)
        else:
            raise

    # Replicate to secondary if requested
    if replicate and self.secondary:
        try:
            self.secondary.save_result(key, result, metadata)
        except Exception as e:
            logger.warning(f"Failed to replicate to secondary storage: {e}")

    # Update cache
    if self.cache:
        try:
            self.cache.save_result(key, result, metadata)
        except Exception as e:
            logger.debug(f"Failed to update cache: {e}")

load_result

load_result(key: str, use_cache: bool = True) -> OptimizationResult

Load result from storage.

Parameters:

Name	Type	Description	Default
key	str	Unique identifier for the result	required
use_cache	bool	Check cache first	True

Returns:

Type	Description
OptimizationResult	OptimizationResult object

Source code in optiscope/storage/manager.py

def load_result(self, key: str, use_cache: bool = True) -> OptimizationResult:
    """
    Load result from storage.

    Args:
        key: Unique identifier for the result
        use_cache: Check cache first

    Returns:
        OptimizationResult object
    """
    # Try cache first
    if use_cache and self.cache and self.cache.exists_result(key):
        try:
            return self.cache.load_result(key)
        except Exception as e:
            logger.debug(f"Cache miss: {e}")

    # Try primary storage
    try:
        result = self.primary.load_result(key)

        # Update cache
        if self.cache:
            try:
                self.cache.save_result(key, result)
            except Exception as e:
                logger.debug(f"Failed to update cache: {e}")

        return result

    except Exception as e:
        logger.warning(f"Failed to load from primary storage: {e}")

        # Try secondary as fallback
        if self.secondary:
            logger.info("Attempting load from secondary storage")
            result = self.secondary.load_result(key)

            # Update cache
            if self.cache:
                try:
                    self.cache.save_result(key, result)
                except Exception as e:
                    logger.debug(f"Failed to update cache: {e}")

            return result
        else:
            raise

delete_result

delete_result(key: str, from_all: bool = True) -> None

Delete result from storage.

Parameters:

Name	Type	Description	Default
key	str	Unique identifier for the result	required
from_all	bool	Delete from all backends (primary, secondary, cache)	True

Source code in optiscope/storage/manager.py

def delete_result(self, key: str, from_all: bool = True) -> None:
    """
    Delete result from storage.

    Args:
        key: Unique identifier for the result
        from_all: Delete from all backends (primary, secondary, cache)
    """
    errors = []

    # Delete from primary
    try:
        self.primary.delete_result(key)
    except Exception as e:
        errors.append(("primary", e))

    # Delete from secondary if requested
    if from_all and self.secondary:
        try:
            self.secondary.delete_result(key)
        except Exception as e:
            errors.append(("secondary", e))

    # Delete from cache
    if from_all and self.cache:
        try:
            self.cache.delete_result(key)
        except Exception as e:
            errors.append(("cache", e))

    if errors:
        logger.warning(f"Errors during deletion: {errors}")

rename_result

rename_result(old_key: str, new_key: str) -> None

Rename a result by changing its key across all configured backends.

Parameters:

Name	Type	Description	Default
old_key	str	The current unique identifier for the result.	required
new_key	str	The new unique identifier for the result.	required

Source code in optiscope/storage/manager.py

def rename_result(self, old_key: str, new_key: str) -> None:
    """
    Rename a result by changing its key across all configured backends.

    Args:
        old_key: The current unique identifier for the result.
        new_key: The new unique identifier for the result.
    """
    if old_key == new_key:
        logger.info(f"Attempted to rename result '{old_key}' to itself. No action taken.")
        return

    logger.info(f"Attempting to rename result from '{old_key}' to '{new_key}'.")

    # 1. Load the result from primary (or secondary if primary fails)
    result = self.load_result(old_key)
    metadata = None
    try:
        metadata = self.primary.get_result_metadata(old_key)
    except Exception as e:
        logger.debug(f"Could not get metadata from primary for '{old_key}': {e}")
        if self.secondary:
            try:
                metadata = self.secondary.get_result_metadata(old_key)
            except Exception as e_sec:
                logger.debug(f"Could not get metadata from secondary for '{old_key}': {e_sec}")

    # 2. Save the result with the new key to all relevant backends
    self.save_result(
        new_key, result, metadata.get("custom") if metadata else None, replicate=True
    )
    logger.debug(f"Result '{old_key}' saved with new key '{new_key}'.")

    # 3. Delete the result with the old key from all backends
    self.delete_result(old_key, from_all=True)
    logger.debug(f"Result '{old_key}' deleted from all backends.")

    logger.info(f"Successfully renamed result from '{old_key}' to '{new_key}'.")

exists_result

exists_result(key: str, check_all: bool = False) -> bool

Check if result exists.

Parameters:

Name	Type	Description	Default
key	str	Unique identifier for the result	required
check_all	bool	Check all backends (otherwise just primary)	False

Returns:

Type	Description
bool	True if result exists

Source code in optiscope/storage/manager.py

def exists_result(self, key: str, check_all: bool = False) -> bool:
    """
    Check if result exists.

    Args:
        key: Unique identifier for the result
        check_all: Check all backends (otherwise just primary)

    Returns:
        True if result exists
    """
    # Check primary
    if self.primary.exists_result(key):
        return True

    # Check secondary if requested
    if check_all and self.secondary and self.secondary.exists_result(key):
        return True

    return False

list_results

list_results(prefix: str | None = None, from_backend: str = 'primary') -> list[str]

List result keys.

Parameters:

Name	Type	Description	Default
prefix	str | None	Optional prefix filter	None
from_backend	str	Which backend to query ("primary", "secondary", "cache")	'primary'

Returns:

Type	Description
list[str]	List of result keys

Source code in optiscope/storage/manager.py

def list_results(self, prefix: str | None = None, from_backend: str = "primary") -> list[str]:
    """
    List result keys.

    Args:
        prefix: Optional prefix filter
        from_backend: Which backend to query ("primary", "secondary", "cache")

    Returns:
        List of result keys
    """
    if from_backend == "primary":
        return self.primary.list_results(prefix)
    elif from_backend == "secondary" and self.secondary:
        return self.secondary.list_results(prefix)
    elif from_backend == "cache" and self.cache:
        return self.cache.list_results(prefix)
    else:
        return []

migrate_result

migrate_result(key: str, from_backend: str, to_backend: str) -> None

Migrate result between backends.

Parameters:

Name	Type	Description	Default
key	str	Result key to migrate	required
from_backend	str	Source backend ("primary", "secondary", "cache")	required
to_backend	str	Destination backend ("primary", "secondary", "cache")	required

Source code in optiscope/storage/manager.py

def migrate_result(self, key: str, from_backend: str, to_backend: str) -> None:
    """
    Migrate result between backends.

    Args:
        key: Result key to migrate
        from_backend: Source backend ("primary", "secondary", "cache")
        to_backend: Destination backend ("primary", "secondary", "cache")
    """
    # Get source and destination
    source = self._get_backend(from_backend)
    destination = self._get_backend(to_backend)

    if source is None or destination is None:
        raise ValueError("Invalid backend specification")

    # Load from source
    result = source.load_result(key)
    metadata = source.get_result_metadata(key)

    # Save to destination
    destination.save_result(key, result, metadata.get("custom"))

    logger.info(f"Migrated '{key}' from {from_backend} to {to_backend}")

sync_backends

sync_backends(direction: str = 'primary_to_secondary') -> dict[str, int]

Synchronize results between backends.

Parameters:

Name	Type	Description	Default
direction	str	Sync direction: - "primary_to_secondary": Copy from primary to secondary - "secondary_to_primary": Copy from secondary to primary - "bidirectional": Two-way sync (newer wins)	'primary_to_secondary'

Returns:

Type	Description
dict[str, int]	Dictionary with sync statistics

Source code in optiscope/storage/manager.py

def sync_backends(self, direction: str = "primary_to_secondary") -> dict[str, int]:
    """
    Synchronize results between backends.

    Args:
        direction: Sync direction:
            - "primary_to_secondary": Copy from primary to secondary
            - "secondary_to_primary": Copy from secondary to primary
            - "bidirectional": Two-way sync (newer wins)

    Returns:
        Dictionary with sync statistics
    """
    if self.secondary is None:
        raise StorageError("Secondary backend not configured")

    stats = {"copied": 0, "skipped": 0, "errors": 0}

    if direction in ["primary_to_secondary", "bidirectional"]:
        # Sync primary -> secondary
        for key in self.primary.list_results():
            try:
                if not self.secondary.exists_result(key):
                    self.migrate_result(key, "primary", "secondary")
                    stats["copied"] += 1
                else:
                    stats["skipped"] += 1
            except Exception as e:
                logger.error(f"Failed to sync '{key}': {e}")
                stats["errors"] += 1

    if direction in ["secondary_to_primary", "bidirectional"]:
        # Sync secondary -> primary
        for key in self.secondary.list_results():
            try:
                if not self.primary.exists_result(key):
                    self.migrate_result(key, "secondary", "primary")
                    stats["copied"] += 1
                else:
                    stats["skipped"] += 1
            except Exception as e:
                logger.error(f"Failed to sync '{key}': {e}")
                stats["errors"] += 1

    return stats

get_storage_info

get_storage_info() -> dict[str, Any]

Get information about all storage backends.

Source code in optiscope/storage/manager.py

def get_storage_info(self) -> dict[str, Any]:
    """Get information about all storage backends."""
    info = {
        "primary": self.primary.get_storage_info(),
    }

    if self.secondary:
        info["secondary"] = self.secondary.get_storage_info()

    if self.cache:
        info["cache"] = self.cache.get_storage_info()

    return info

clear_cache

clear_cache() -> int

Clear cache.

Source code in optiscope/storage/manager.py

def clear_cache(self) -> int:
    """Clear cache."""
    if self.cache:
        return self.cache.clear()
    return 0

clear_all_results

clear_all_results(from_all: bool = True) -> None

Clear all results from storage.

Parameters:

Name	Type	Description	Default
from_all	bool	Clear from all backends (primary, secondary, cache)	True

Source code in optiscope/storage/manager.py

def clear_all_results(self, from_all: bool = True) -> None:
    """
    Clear all results from storage.

    Args:
        from_all: Clear from all backends (primary, secondary, cache)
    """
    keys_to_delete = self.list_results(from_backend="primary")
    for key in keys_to_delete:
        self.delete_result(key, from_all=from_all)
    logger.info(f"Cleared {len(keys_to_delete)} results from primary storage.")

    if from_all and self.secondary:
        secondary_keys = self.list_results(from_backend="secondary")
        for key in secondary_keys:
            self.secondary.delete_result(key)
        logger.info(f"Cleared {len(secondary_keys)} results from secondary storage.")

    if from_all and self.cache:
        self.cache.clear()
        logger.info("Cleared cache.")

close

close() -> None

Close all storage backends.

Source code in optiscope/storage/manager.py

def close(self) -> None:
    """Close all storage backends."""
    self.primary.close()

    if self.secondary:
        self.secondary.close()

    if self.cache:
        self.cache.close()

get_result_storage_type

get_result_storage_type(key: str) -> StorageType | None

Determine the storage type for a given result key.

Parameters:

Name	Type	Description	Default
key	str	Unique identifier for the result.	required

Returns:

Type	Description
StorageType | None	The StorageType of the backend where the result is found, or None if not found.

Source code in optiscope/storage/manager.py

def get_result_storage_type(self, key: str) -> StorageType | None:
    """
    Determine the storage type for a given result key.

    Args:
        key: Unique identifier for the result.

    Returns:
        The StorageType of the backend where the result is found, or None if not found.
    """
    if self.primary.exists_result(key):
        return self.primary.storage_type
    if self.secondary and self.secondary.exists_result(key):
        return self.secondary.storage_type
    if self.cache and self.cache.exists_result(key):
        return self.cache.storage_type
    return None

__enter__

__enter__() -> StorageManager

Context manager entry.

Source code in optiscope/storage/manager.py

def __enter__(self) -> StorageManager:
    """Context manager entry."""
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb) -> None

Context manager exit.

Source code in optiscope/storage/manager.py

def __exit__(self, exc_type, exc_val, exc_tb) -> None:
    """Context manager exit."""
    self.close()

to_dict

to_dict() -> dict[str, Any]

Serialize the storage manager to a dictionary.

Source code in optiscope/storage/manager.py

def to_dict(self) -> dict[str, Any]:
    """Serialize the storage manager to a dictionary."""
    # For persistent storage backends (database, filesystem),
    # don't serialize the cache contents to avoid browser storage quota issues.
    # The cache can be rebuilt on load from the persistent storage.
    cache_data = None
    if self.cache and self.primary.storage_type == StorageType.MEMORY:
        # Only serialize cache if primary storage is memory
        cache_data = self.cache.to_dict()
    elif self.cache:
        # For persistent backends, just store cache configuration
        cache_data = {
            "storage_type": "memory",
            "results": {},  # Empty results to avoid quota issues
            "config": self.cache.config,
        }

    return {
        "primary": self.primary.to_dict(),
        "secondary": self.secondary.to_dict() if self.secondary else None,
        "use_cache": self.use_cache,
        "cache": cache_data,
    }

from_dict classmethod

from_dict(data: dict[str, Any]) -> StorageManager

Deserialize a storage manager from a dictionary.

Source code in optiscope/storage/manager.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> StorageManager:
    """Deserialize a storage manager from a dictionary."""

    def restore_backend(backend_data: dict | None) -> StorageBackend | None:
        if not backend_data:
            return None

        storage_type = backend_data.get("storage_type")
        if storage_type == "memory":
            return MemoryStorage.from_dict(backend_data)
        elif storage_type == "filesystem":
            return FileSystemStorage.from_dict(backend_data)
        elif storage_type == "database":
            if not DATABASE_AVAILABLE:
                raise StorageError(
                    "Database backend not available. Install with: pip install optiscope[database]"
                )
            return DatabaseStorage.from_dict(backend_data)
        # Add other storage types here as they become serializable
        raise ValueError(f"Unknown storage type for deserialization: {storage_type}")

    primary = restore_backend(data["primary"])
    if not primary:
        raise ValueError("Primary storage backend could not be deserialized.")

    secondary = restore_backend(data.get("secondary"))

    manager = cls(primary=primary, secondary=secondary, use_cache=data.get("use_cache", True))

    if "cache" in data and data["cache"]:
        manager.cache = MemoryStorage.from_dict(data["cache"])

    return manager

Functions

create_memory_storage

create_memory_storage(**kwargs: Any) -> MemoryStorage

Create memory storage backend.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Configuration options	{}

Returns:

Type	Description
MemoryStorage	MemoryStorage instance

Source code in optiscope/storage/manager.py

def create_memory_storage(**kwargs: Any) -> MemoryStorage:
    """
    Create memory storage backend.

    Args:
        **kwargs: Configuration options

    Returns:
        MemoryStorage instance
    """
    return MemoryStorage(**kwargs)

create_filesystem_storage

create_filesystem_storage(base_path: str | Path, **kwargs: Any) -> FileSystemStorage

Create filesystem storage backend.

Parameters:

Name	Type	Description	Default
base_path	str | Path	Base directory for storage	required
**kwargs	Any	Configuration options	{}

Returns:

Type	Description
FileSystemStorage	FileSystemStorage instance

Source code in optiscope/storage/manager.py

def create_filesystem_storage(base_path: str | Path, **kwargs: Any) -> FileSystemStorage:
    """
    Create filesystem storage backend.

    Args:
        base_path: Base directory for storage
        **kwargs: Configuration options

    Returns:
        FileSystemStorage instance
    """
    return FileSystemStorage(base_path=base_path, **kwargs)

create_database_storage

create_database_storage(connection_string: str, **kwargs: Any) -> DatabaseStorage

Create database storage backend.

Parameters:

Name	Type	Description	Default
connection_string	str	Database connection string	required
**kwargs	Any	Configuration options	{}

Returns:

Type	Description
DatabaseStorage	DatabaseStorage instance

Raises:

Type	Description
StorageError	If database backend not available

Source code in optiscope/storage/manager.py

def create_database_storage(connection_string: str, **kwargs: Any) -> DatabaseStorage:
    """
    Create database storage backend.

    Args:
        connection_string: Database connection string
        **kwargs: Configuration options

    Returns:
        DatabaseStorage instance

    Raises:
        StorageError: If database backend not available
    """
    if not DATABASE_AVAILABLE:
        raise StorageError(
            "Database backend not available. Install with: pip install optiscope[database]"
        )

    from optiscope.storage.database import DatabaseStorage

    return DatabaseStorage(connection_string=connection_string, **kwargs)

create_storage_manager

create_storage_manager(primary_type: str = 'memory', primary_config: dict[str, Any] | None = None, secondary_type: str | None = None, secondary_config: dict[str, Any] | None = None, **manager_kwargs: Any) -> StorageManager

Create storage manager with specified backends.

Parameters:

Name	Type	Description	Default
primary_type	str	Type of primary backend ("memory", "filesystem", "database")	'memory'
primary_config	dict[str, Any] | None	Configuration for primary backend	None
secondary_type	str | None	Optional type of secondary backend	None
secondary_config	dict[str, Any] | None	Configuration for secondary backend	None
**manager_kwargs	Any	Additional manager configuration	{}

Returns:

Type	Description
StorageManager	StorageManager instance

Source code in optiscope/storage/manager.py

def create_storage_manager(
    primary_type: str = "memory",
    primary_config: dict[str, Any] | None = None,
    secondary_type: str | None = None,
    secondary_config: dict[str, Any] | None = None,
    **manager_kwargs: Any,
) -> StorageManager:
    """
    Create storage manager with specified backends.

    Args:
        primary_type: Type of primary backend ("memory", "filesystem", "database")
        primary_config: Configuration for primary backend
        secondary_type: Optional type of secondary backend
        secondary_config: Configuration for secondary backend
        **manager_kwargs: Additional manager configuration

    Returns:
        StorageManager instance
    """
    primary_config = primary_config or {}

    # Create primary backend
    if primary_type == "memory":
        primary = create_memory_storage(**primary_config)
    elif primary_type == "filesystem":
        primary = create_filesystem_storage(**primary_config)
    elif primary_type == "database":
        primary = create_database_storage(**primary_config)
    else:
        raise ValueError(f"Unknown storage type: {primary_type}")

    # Create secondary backend if specified
    secondary = None
    if secondary_type:
        secondary_config = secondary_config or {}

        if secondary_type == "memory":
            secondary = create_memory_storage(**secondary_config)
        elif secondary_type == "filesystem":
            secondary = create_filesystem_storage(**secondary_config)
        elif secondary_type == "database":
            secondary = create_database_storage(**secondary_config)
        else:
            raise ValueError(f"Unknown storage type: {secondary_type}")

    return StorageManager(primary=primary, secondary=secondary, **manager_kwargs)