Storage Backends

Base Storage

base

Base storage abstraction for optimization results.

This module defines the abstract interface for storage backends, enabling different storage strategies (memory, filesystem, database) with a consistent API.

Classes

StorageType

Bases: str, Enum

Types of storage backends.

StorageBackend

StorageBackend(**config: Any)

Bases: ABC

Abstract base class for storage backends.

Storage backends handle persistence of OptimizationResult objects and their associated result sets. Each backend can implement its own strategy for data storage and retrieval.

Initialize storage backend.

Parameters:

Name	Type	Description	Default
**config	Any	Backend-specific configuration	{}

Source code in optiscope/storage/base.py

def __init__(self, **config: Any) -> None:
    """
    Initialize storage backend.

    Args:
        **config: Backend-specific configuration
    """
    self.config = config
    self._initialize()

Functions

save_result abstractmethod

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

Save an optimization result.

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 additional metadata	None

Raises:

Type	Description
StorageError	If save operation fails

Source code in optiscope/storage/base.py

@abstractmethod
def save_result(
    self, key: str, result: OptimizationResult, metadata: dict[str, Any] | None = None
) -> None:
    """
    Save an optimization result.

    Args:
        key: Unique identifier for the result
        result: OptimizationResult to save
        metadata: Optional additional metadata

    Raises:
        StorageError: If save operation fails
    """
    pass

load_result abstractmethod

load_result(key: str) -> OptimizationResult

Load an optimization result.

Parameters:

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

Returns:

Type	Description
OptimizationResult	OptimizationResult object

Raises:

Type	Description
KeyError	If key not found
StorageError	If load operation fails

Source code in optiscope/storage/base.py

@abstractmethod
def load_result(self, key: str) -> OptimizationResult:
    """
    Load an optimization result.

    Args:
        key: Unique identifier for the result

    Returns:
        OptimizationResult object

    Raises:
        KeyError: If key not found
        StorageError: If load operation fails
    """
    pass

delete_result abstractmethod

delete_result(key: str) -> None

Delete an optimization result.

Parameters:

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

Raises:

Type	Description
KeyError	If key not found
StorageError	If delete operation fails

Source code in optiscope/storage/base.py

@abstractmethod
def delete_result(self, key: str) -> None:
    """
    Delete an optimization result.

    Args:
        key: Unique identifier for the result

    Raises:
        KeyError: If key not found
        StorageError: If delete operation fails
    """
    pass

exists_result abstractmethod

exists_result(key: str) -> bool

Check if a result exists.

Parameters:

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

Returns:

Type	Description
bool	True if result exists

Source code in optiscope/storage/base.py

@abstractmethod
def exists_result(self, key: str) -> bool:
    """
    Check if a result exists.

    Args:
        key: Unique identifier for the result

    Returns:
        True if result exists
    """
    pass

list_results abstractmethod

list_results(prefix: str | None = None) -> list[str]

List all stored result keys.

Parameters:

Name	Type	Description	Default
prefix	str | None	Optional prefix filter	None

Returns:

Type	Description
list[str]	List of result keys

Source code in optiscope/storage/base.py

@abstractmethod
def list_results(self, prefix: str | None = None) -> list[str]:
    """
    List all stored result keys.

    Args:
        prefix: Optional prefix filter

    Returns:
        List of result keys
    """
    pass

rename_result abstractmethod

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

Rename a result.

Parameters:

Name	Type	Description	Default
old_key	str	The current key of the result	required
new_key	str	The new key for the result	required

Raises:

Type	Description
KeyError	If old_key not found
ValueError	If new_key already exists
StorageError	If rename operation fails

Source code in optiscope/storage/base.py

@abstractmethod
def rename_result(self, old_key: str, new_key: str) -> None:
    """
    Rename a result.

    Args:
        old_key: The current key of the result
        new_key: The new key for the result

    Raises:
        KeyError: If old_key not found
        ValueError: If new_key already exists
        StorageError: If rename operation fails
    """
    pass

save_set abstractmethod

save_set(result_key: str, set_name: str, result_set: ResultSet) -> None

Save a result set for a specific result.

Parameters:

Name	Type	Description	Default
result_key	str	Key of the parent result	required
set_name	str	Name of the result set	required
result_set	ResultSet	ResultSet to save	required

Raises:

Type	Description
KeyError	If result_key not found
StorageError	If save operation fails

Source code in optiscope/storage/base.py

@abstractmethod
def save_set(self, result_key: str, set_name: str, result_set: ResultSet) -> None:
    """
    Save a result set for a specific result.

    Args:
        result_key: Key of the parent result
        set_name: Name of the result set
        result_set: ResultSet to save

    Raises:
        KeyError: If result_key not found
        StorageError: If save operation fails
    """
    pass

load_set abstractmethod

load_set(result_key: str, set_name: str) -> ResultSet

Load a result set.

Parameters:

Name	Type	Description	Default
result_key	str	Key of the parent result	required
set_name	str	Name of the result set	required

Returns:

Type	Description
ResultSet	ResultSet object

Raises:

Type	Description
KeyError	If result_key or set_name not found
StorageError	If load operation fails

Source code in optiscope/storage/base.py

@abstractmethod
def load_set(self, result_key: str, set_name: str) -> ResultSet:
    """
    Load a result set.

    Args:
        result_key: Key of the parent result
        set_name: Name of the result set

    Returns:
        ResultSet object

    Raises:
        KeyError: If result_key or set_name not found
        StorageError: If load operation fails
    """
    pass

delete_set abstractmethod

delete_set(result_key: str, set_name: str) -> None

Delete a result set.

Parameters:

Name	Type	Description	Default
result_key	str	Key of the parent result	required
set_name	str	Name of the result set	required

Raises:

Type	Description
KeyError	If result_key or set_name not found
StorageError	If delete operation fails

Source code in optiscope/storage/base.py

@abstractmethod
def delete_set(self, result_key: str, set_name: str) -> None:
    """
    Delete a result set.

    Args:
        result_key: Key of the parent result
        set_name: Name of the result set

    Raises:
        KeyError: If result_key or set_name not found
        StorageError: If delete operation fails
    """
    pass

list_sets abstractmethod

list_sets(result_key: str) -> list[str]

List all result sets for a given result.

Parameters:

Name	Type	Description	Default
result_key	str	Key of the parent result	required

Returns:

Type	Description
list[str]	List of set names

Raises:

Type	Description
KeyError	If result_key not found

Source code in optiscope/storage/base.py

@abstractmethod
def list_sets(self, result_key: str) -> list[str]:
    """
    List all result sets for a given result.

    Args:
        result_key: Key of the parent result

    Returns:
        List of set names

    Raises:
        KeyError: If result_key not found
    """
    pass

get_result_metadata abstractmethod

get_result_metadata(key: str) -> dict[str, Any]

Get metadata for a result without loading the full data.

Parameters:

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

Returns:

Type	Description
dict[str, Any]	Dictionary of metadata

Raises:

Type	Description
KeyError	If key not found

Source code in optiscope/storage/base.py

@abstractmethod
def get_result_metadata(self, key: str) -> dict[str, Any]:
    """
    Get metadata for a result without loading the full data.

    Args:
        key: Unique identifier for the result

    Returns:
        Dictionary of metadata

    Raises:
        KeyError: If key not found
    """
    pass

update_result_metadata abstractmethod

update_result_metadata(key: str, metadata: dict[str, Any]) -> None

Update metadata for a result.

Parameters:

Name	Type	Description	Default
key	str	Unique identifier for the result	required
metadata	dict[str, Any]	Metadata to update/add	required

Raises:

Type	Description
KeyError	If key not found

Source code in optiscope/storage/base.py

@abstractmethod
def update_result_metadata(self, key: str, metadata: dict[str, Any]) -> None:
    """
    Update metadata for a result.

    Args:
        key: Unique identifier for the result
        metadata: Metadata to update/add

    Raises:
        KeyError: If key not found
    """
    pass

clear abstractmethod

clear(prefix: str | None = None) -> int

Clear stored results.

Parameters:

Name	Type	Description	Default
prefix	str | None	Optional prefix filter (if None, clear all)	None

Returns:

Type	Description
int	Number of results deleted

Source code in optiscope/storage/base.py

@abstractmethod
def clear(self, prefix: str | None = None) -> int:
    """
    Clear stored results.

    Args:
        prefix: Optional prefix filter (if None, clear all)

    Returns:
        Number of results deleted
    """
    pass

get_storage_info abstractmethod

get_storage_info() -> dict[str, Any]

Get information about storage usage.

Returns:

Type	Description
dict[str, Any]	Dictionary with storage information

Source code in optiscope/storage/base.py

@abstractmethod
def get_storage_info(self) -> dict[str, Any]:
    """
    Get information about storage usage.

    Returns:
        Dictionary with storage information
    """
    pass

to_dict abstractmethod

to_dict() -> dict[str, Any]

Serialize the storage backend's state to a dictionary.

For memory-based storage, this might involve serializing all contained data. For file-based or database storage, this might involve storing references (e.g., file paths or connection strings) so the state can be reconstructed.

Source code in optiscope/storage/base.py

@abstractmethod
def to_dict(self) -> dict[str, Any]:
    """
    Serialize the storage backend's state to a dictionary.

    For memory-based storage, this might involve serializing all contained data.
    For file-based or database storage, this might involve storing references
    (e.g., file paths or connection strings) so the state can be reconstructed.
    """
    pass

from_dict abstractmethod classmethod

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

Deserialize a storage backend from a dictionary representation.

This class method should reconstruct the storage backend from the serialized state produced by to_dict.

Source code in optiscope/storage/base.py

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

    This class method should reconstruct the storage backend from the serialized
    state produced by `to_dict`.
    """
    pass

close

close() -> None

Close storage backend (cleanup connections, etc.).

Source code in optiscope/storage/base.py

def close(self) -> None:
    """Close storage backend (cleanup connections, etc.)."""
    pass

__enter__

__enter__() -> StorageBackend

Context manager entry.

Source code in optiscope/storage/base.py

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

__exit__

__exit__(exc_type, exc_val, exc_tb) -> None

Context manager exit.

Source code in optiscope/storage/base.py

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

StorageError

Bases: Exception

Base exception for storage-related errors.

StorageConnectionError

Bases: StorageError

Exception raised when storage connection fails.

StorageOperationError

Bases: StorageError

Exception raised when storage operation fails.

StorageCapacityError

Bases: StorageError

Exception raised when storage capacity is exceeded.

StorageMetadata

StorageMetadata(key: str, created_at: datetime, modified_at: datetime, size_bytes: int | None = None, n_points: int | None = None, n_sets: int | None = None, custom_metadata: dict[str, Any] | None = None)

Metadata about a stored result.

Initialize storage metadata.

Parameters:

Name	Type	Description	Default
key	str	Result key	required
created_at	datetime	Creation timestamp	required
modified_at	datetime	Last modification timestamp	required
size_bytes	int | None	Size in bytes	None
n_points	int | None	Number of data points	None
n_sets	int | None	Number of result sets	None
custom_metadata	dict[str, Any] | None	Additional custom metadata	None

Source code in optiscope/storage/base.py

def __init__(
    self,
    key: str,
    created_at: datetime,
    modified_at: datetime,
    size_bytes: int | None = None,
    n_points: int | None = None,
    n_sets: int | None = None,
    custom_metadata: dict[str, Any] | None = None,
) -> None:
    """
    Initialize storage metadata.

    Args:
        key: Result key
        created_at: Creation timestamp
        modified_at: Last modification timestamp
        size_bytes: Size in bytes
        n_points: Number of data points
        n_sets: Number of result sets
        custom_metadata: Additional custom metadata
    """
    self.key = key
    self.created_at = created_at
    self.modified_at = modified_at
    self.size_bytes = size_bytes
    self.n_points = n_points
    self.n_sets = n_sets
    self.custom_metadata = custom_metadata or {}

Functions

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

Source code in optiscope/storage/base.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary."""
    return {
        "key": self.key,
        "created_at": self.created_at.isoformat(),
        "modified_at": self.modified_at.isoformat(),
        "size_bytes": self.size_bytes,
        "n_points": self.n_points,
        "n_sets": self.n_sets,
        "custom_metadata": self.custom_metadata,
    }

from_dict classmethod

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

Create from dictionary.

Source code in optiscope/storage/base.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> StorageMetadata:
    """Create from dictionary."""
    return cls(
        key=data["key"],
        created_at=datetime.fromisoformat(data["created_at"]),
        modified_at=datetime.fromisoformat(data["modified_at"]),
        size_bytes=data.get("size_bytes"),
        n_points=data.get("n_points"),
        n_sets=data.get("n_sets"),
        custom_metadata=data.get("custom_metadata"),
    )

Database Storage

database

Database storage backend for optimization results.

Scalable storage backend using SQLAlchemy for database operations. Supports PostgreSQL, SQLite, MySQL, and other SQLAlchemy-compatible databases.

Classes

DatabaseStorage

DatabaseStorage(connection_string: str, echo: bool = False, pool_size: int = 5, max_overflow: int = 10, dump_as_json: bool = False, **config: Any)

Bases: StorageBackend

Database storage backend using SQLAlchemy.

Stores results in a relational database with efficient querying and metadata indexing. Suitable for large-scale applications and multi-user scenarios.

For multi-source session databases, use SessionDBStorage instead, which adds automatic source_id prefixing for data isolation.

Initialize database storage.

Parameters:

Name	Type	Description	Default
connection_string	str	SQLAlchemy connection string Examples: - SQLite: "sqlite:///path/to/database.db" - PostgreSQL: "postgresql://user:pass@localhost/dbname" - MySQL: "mysql://user:pass@localhost/dbname" - DuckDB: "duckdb:///database.duckdb"	required
echo	bool	Enable SQL query logging	False
pool_size	int	Connection pool size	5
max_overflow	int	Maximum overflow connections	10
**config	Any	Additional SQLAlchemy engine configuration	{}

Source code in optiscope/storage/database.py

def __init__(
    self,
    connection_string: str,
    echo: bool = False,
    pool_size: int = 5,
    max_overflow: int = 10,
    dump_as_json: bool = False,
    **config: Any,
) -> None:
    """
    Initialize database storage.

    Args:
        connection_string: SQLAlchemy connection string
            Examples:
            - SQLite: "sqlite:///path/to/database.db"
            - PostgreSQL: "postgresql://user:pass@localhost/dbname"
            - MySQL: "mysql://user:pass@localhost/dbname"
            - DuckDB: "duckdb:///database.duckdb"
        echo: Enable SQL query logging
        pool_size: Connection pool size
        max_overflow: Maximum overflow connections
        **config: Additional SQLAlchemy engine configuration
    """
    if not SQLALCHEMY_AVAILABLE:
        raise StorageError(
            "SQLAlchemy is not installed. Install with: pip install optiscope[database]"
        )

    self.connection_string = connection_string
    self.echo = echo
    self.pool_size = pool_size
    self.max_overflow = max_overflow
    self.dump_as_json = dump_as_json
    # Create backend-specific ResultRecord and SetRecord classes
    self.ResultRecord, self.SetRecord = _create_set_record_classes(
        connection_string, dump_as_json
    )

    super().__init__(**config)

Functions

save_result

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

Save result to database.

Source code in optiscope/storage/database.py

def save_result(
    self, key: str, result: OptimizationResult, metadata: dict[str, Any] | None = None
) -> None:
    """Save result to database."""
    session = self.SessionLocal()

    try:
        # Serialize result
        data = self._serialize_result(result)
        if isinstance(data, str) and self.dump_as_json:
            size_bytes = len(data.encode("utf-8"))
        else:
            size_bytes = len(data)

        # Check if record exists
        existing: Any = session.query(self.ResultRecord).filter_by(key=key).first()

        now = datetime.now()

        if existing:
            # Update existing record
            existing.data = data
            existing.modified_at = now
            existing.n_points = result.n_points
            existing.n_sets = len(result.list_sets())
            existing.size_bytes = size_bytes

            # Update custom metadata
            if metadata:
                existing_meta = json.loads(existing.metadata_json or "{}")
                existing_meta.update(metadata)
                existing.metadata_json = json.dumps(existing_meta)
        else:
            # Create new record
            record: Any = self.ResultRecord(
                key=key,
                data=data,
                created_at=now,
                modified_at=now,
                n_points=result.n_points,
                n_sets=len(result.list_sets()),
                size_bytes=size_bytes,
                metadata_json=json.dumps(metadata or {}),
            )
            session.add(record)

        session.commit()

        # Save result sets
        for set_name, result_set in result.sets.items():
            self.save_set(key, set_name, result_set)

    except Exception as e:
        session.rollback()
        raise StorageOperationError(f"Failed to save result '{key}': {str(e)}") from e
    finally:
        session.close()

load_result

load_result(key: str) -> OptimizationResult

Load result from database.

Source code in optiscope/storage/database.py

def load_result(self, key: str) -> OptimizationResult:
    """Load result from database."""
    session = self.SessionLocal()

    try:
        record: Any = session.query(self.ResultRecord).filter_by(key=key).first()

        if not record:
            raise KeyError(f"Result '{key}' not found in storage")

        # Deserialize result
        result = self._deserialize_result(record.data)

        # Load result sets
        set_records: list[Any] = session.query(self.SetRecord).filter_by(result_key=key).all()
        for set_record in set_records:
            try:
                set_data = json.loads(set_record.data)
                result_set = ResultSet.from_dict(set_data)
                result._set_manager.add_set(result_set)
            except Exception as e:
                print(f"Warning: Failed to load set '{set_record.set_name}': {e}")

        return result

    except KeyError:
        raise
    except Exception as e:
        raise StorageOperationError(f"Failed to load result '{key}': {str(e)}") from e
    finally:
        session.close()

delete_result

delete_result(key: str) -> None

Delete result from database.

Source code in optiscope/storage/database.py

def delete_result(self, key: str) -> None:
    """Delete result from database."""
    session = self.SessionLocal()

    try:
        record: Any = session.query(self.ResultRecord).filter_by(key=key).first()

        if not record:
            raise KeyError(f"Result '{key}' not found in storage")

        # Delete result sets first
        session.query(self.SetRecord).filter_by(result_key=key).delete()

        # Delete result
        session.delete(record)
        session.commit()

    except KeyError:
        raise
    except Exception as e:
        session.rollback()
        raise StorageOperationError(f"Failed to delete result '{key}': {str(e)}") from e
    finally:
        session.close()

rename_result

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

Rename a result.

Parameters:

Name	Type	Description	Default
old_key	str	Current result key	required
new_key	str	New result key	required

Source code in optiscope/storage/database.py

def rename_result(self, old_key: str, new_key: str) -> None:
    """
    Rename a result.

    Args:
        old_key: Current result key
        new_key: New result key
    """
    session = self.SessionLocal()

    try:
        # Check if new key exists
        if session.query(self.ResultRecord).filter_by(key=new_key).count() > 0:
            raise StorageError(f"Result '{new_key}' already exists")

        # Get existing record
        record: Any = session.query(self.ResultRecord).filter_by(key=old_key).first()

        if not record:
            raise KeyError(f"Result '{old_key}' not found in storage")

        # Update key
        record.key = new_key
        record.modified_at = datetime.now()

        # Update serialized data
        result = self._deserialize_result(record.data)
        result.rename(new_key)
        record.data = self._serialize_result(result)

        # Update size_bytes if needed
        if isinstance(record.data, str) and self.dump_as_json:
            record.size_bytes = len(record.data.encode("utf-8"))
        else:
            record.size_bytes = len(record.data)

        # Update SetRecords associated with this result
        session.query(self.SetRecord).filter_by(result_key=old_key).update(
            {"result_key": new_key}
        )

        session.commit()

    except (KeyError, StorageError):
        session.rollback()
        raise
    except Exception as e:
        session.rollback()
        raise StorageOperationError(
            f"Failed to rename result '{old_key}' to '{new_key}': {str(e)}"
        ) from e
    finally:
        session.close()

exists_result

exists_result(key: str) -> bool

Check if result exists.

Source code in optiscope/storage/database.py

def exists_result(self, key: str) -> bool:
    """Check if result exists."""
    session = self.SessionLocal()
    try:
        count = session.query(self.ResultRecord).filter_by(key=key).count()
        return count > 0
    finally:
        session.close()

list_results

list_results(prefix: str | None = None) -> list[str]

List all result keys.

Source code in optiscope/storage/database.py

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

    try:
        query = session.query(self.ResultRecord.key)

        if prefix:
            query = query.filter(self.ResultRecord.key.like(f"{prefix}%"))

        keys = [row[0] for row in query.all()]
        return sorted(keys)
    finally:
        session.close()

save_set

save_set(result_key: str, set_name: str, result_set: ResultSet) -> None

Save result set to database.

Source code in optiscope/storage/database.py

def save_set(self, result_key: str, set_name: str, result_set: ResultSet) -> None:
    """Save result set to database."""
    session = self.SessionLocal()

    try:
        # Check if result exists
        if not self.exists_result(result_key):
            raise KeyError(f"Result '{result_key}' not found in storage")

        # Check if set record exists
        existing: Any = (
            session.query(self.SetRecord)
            .filter_by(result_key=result_key, set_name=set_name)
            .first()
        )

        now = datetime.now()
        set_data = json.dumps(result_set.to_dict(), default=str)

        if existing:
            # Update existing record
            existing.data = set_data
            existing.modified_at = now
        else:
            # Create new record
            record: Any = self.SetRecord(
                result_key=result_key,
                set_name=set_name,
                data=set_data,
                created_at=now,
                modified_at=now,
            )
            session.add(record)

        session.commit()

        # Update result metadata
        result_record: Any = session.query(self.ResultRecord).filter_by(key=result_key).first()
        if result_record:
            result_record.modified_at = now
            result_record.n_sets = (
                session.query(self.SetRecord).filter_by(result_key=result_key).count()
            )
            session.commit()

    except KeyError:
        raise
    except Exception as e:
        session.rollback()
        raise StorageOperationError(f"Failed to save set '{set_name}': {str(e)}") from e
    finally:
        session.close()

load_set

load_set(result_key: str, set_name: str) -> ResultSet

Load result set from database.

Source code in optiscope/storage/database.py

def load_set(self, result_key: str, set_name: str) -> ResultSet:
    """Load result set from database."""
    session = self.SessionLocal()

    try:
        record: Any = (
            session.query(self.SetRecord)
            .filter_by(result_key=result_key, set_name=set_name)
            .first()
        )

        if not record:
            raise KeyError(f"Set '{set_name}' not found for result '{result_key}'")

        set_data = json.loads(record.data)
        return ResultSet.from_dict(set_data)

    except KeyError:
        raise
    except Exception as e:
        raise StorageOperationError(f"Failed to load set '{set_name}': {str(e)}") from e
    finally:
        session.close()

delete_set

delete_set(result_key: str, set_name: str) -> None

Delete result set from database.

Source code in optiscope/storage/database.py

def delete_set(self, result_key: str, set_name: str) -> None:
    """Delete result set from database."""
    session = self.SessionLocal()

    try:
        record: Any = (
            session.query(self.SetRecord)
            .filter_by(result_key=result_key, set_name=set_name)
            .first()
        )

        if not record:
            raise KeyError(f"Set '{set_name}' not found for result '{result_key}'")

        session.delete(record)
        session.commit()

        # Update result metadata
        result_record: Any = session.query(self.ResultRecord).filter_by(key=result_key).first()
        if result_record:
            result_record.modified_at = datetime.now()
            result_record.n_sets = (
                session.query(self.SetRecord).filter_by(result_key=result_key).count()
            )
            session.commit()

    except KeyError:
        raise
    except Exception as e:
        session.rollback()
        raise StorageOperationError(f"Failed to delete set '{set_name}': {str(e)}") from e
    finally:
        session.close()

list_sets

list_sets(result_key: str) -> list[str]

List all sets for a result.

Source code in optiscope/storage/database.py

def list_sets(self, result_key: str) -> list[str]:
    """List all sets for a result."""
    session = self.SessionLocal()

    try:
        if not self.exists_result(result_key):
            raise KeyError(f"Result '{result_key}' not found in storage")

        records = session.query(self.SetRecord.set_name).filter_by(result_key=result_key).all()

        return sorted([row[0] for row in records])
    finally:
        session.close()

get_result_metadata

get_result_metadata(key: str) -> dict[str, Any]

Get result metadata.

Source code in optiscope/storage/database.py

def get_result_metadata(self, key: str) -> dict[str, Any]:
    """Get result metadata."""
    session = self.SessionLocal()

    try:
        record: Any = session.query(self.ResultRecord).filter_by(key=key).first()

        if not record:
            raise KeyError(f"Result '{key}' not found in storage")

        custom_meta = json.loads(record.metadata_json or "{}")

        return {
            "key": record.key,
            "created_at": record.created_at,
            "modified_at": record.modified_at,
            "n_points": record.n_points,
            "n_sets": record.n_sets,
            "size_bytes": record.size_bytes,
            "custom": custom_meta,
        }
    finally:
        session.close()

update_result_metadata

update_result_metadata(key: str, metadata: dict[str, Any]) -> None

Update result metadata.

Source code in optiscope/storage/database.py

def update_result_metadata(self, key: str, metadata: dict[str, Any]) -> None:
    """Update result metadata."""
    session = self.SessionLocal()

    try:
        record: Any = session.query(self.ResultRecord).filter_by(key=key).first()

        if not record:
            raise KeyError(f"Result '{key}' not found in storage")

        # Update custom metadata
        existing_meta = json.loads(record.metadata_json or "{}")
        existing_meta.update(metadata)
        record.metadata_json = json.dumps(existing_meta)
        record.modified_at = datetime.now()

        session.commit()

    except KeyError:
        raise
    except Exception as e:
        session.rollback()
        raise StorageOperationError(f"Failed to update metadata for '{key}': {str(e)}") from e
    finally:
        session.close()

clear

clear(prefix: str | None = None) -> int

Clear stored results.

Source code in optiscope/storage/database.py

def clear(self, prefix: str | None = None) -> int:
    """Clear stored results."""
    session = self.SessionLocal()

    try:
        query = session.query(self.ResultRecord)

        if prefix:
            query = query.filter(self.ResultRecord.key.like(f"{prefix}%"))

        # Get keys to delete sets
        results: list[Any] = query.all()
        keys = [row.key for row in results]

        # Delete sets first
        for key in keys:
            session.query(self.SetRecord).filter_by(result_key=key).delete()

        # Delete results
        count = query.delete()
        session.commit()

        return count

    except Exception as e:
        session.rollback()
        raise StorageOperationError(f"Failed to clear storage: {str(e)}") from e
    finally:
        session.close()

get_storage_info

get_storage_info() -> dict[str, Any]

Get storage information.

Source code in optiscope/storage/database.py

def get_storage_info(self) -> dict[str, Any]:
    """Get storage information."""
    session = self.SessionLocal()

    try:
        n_results = session.query(self.ResultRecord).count()
        n_sets = session.query(self.SetRecord).count()

        # Get total size
        total_size = session.query(self.ResultRecord.size_bytes).all()
        total_mb = sum(s[0] for s in total_size if s[0]) / (1024 * 1024)

        return {
            "storage_type": "database",
            "connection_string": self.connection_string.split("@")[-1],  # Hide credentials
            "n_results": n_results,
            "n_total_sets": n_sets,
            "total_size_mb": total_mb,
        }
    finally:
        session.close()

close

close() -> None

Close database connections.

Source code in optiscope/storage/database.py

def close(self) -> None:
    """Close database connections."""
    if hasattr(self, "engine"):
        self.engine.dispose()

vacuum

vacuum() -> None

Optimize database (SQLite only).

For SQLite databases, runs VACUUM to reclaim space. For other databases, this is a no-op.

Source code in optiscope/storage/database.py

def vacuum(self) -> None:
    """
    Optimize database (SQLite only).

    For SQLite databases, runs VACUUM to reclaim space.
    For other databases, this is a no-op.
    """
    if "sqlite" in self.connection_string.lower():
        session = self.SessionLocal()
        try:
            session.execute(text("VACUUM"))
            session.commit()
        finally:
            session.close()

to_dict

to_dict() -> dict[str, Any]

Serialize database storage to dictionary.

Note: This only stores connection information, not the actual data. The data persists in the database itself.

Source code in optiscope/storage/database.py

def to_dict(self) -> dict[str, Any]:
    """
    Serialize database storage to dictionary.

    Note: This only stores connection information, not the actual data.
    The data persists in the database itself.
    """
    return {
        "storage_type": "database",
        "connection_string": self.connection_string,
        "echo": self.echo,
        "pool_size": self.pool_size,
        "max_overflow": self.max_overflow,
        "config": self.config,
    }

from_dict classmethod

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

Deserialize database storage from dictionary.

Reconnects to the database using the stored connection information.

Source code in optiscope/storage/database.py

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

    Reconnects to the database using the stored connection information.
    """
    return cls(
        connection_string=data["connection_string"],
        echo=data.get("echo", False),
        pool_size=data.get("pool_size", 5),
        max_overflow=data.get("max_overflow", 10),
        **data.get("config", {}),
    )

File System Storage

file_system

File system storage backend for optimization results.

Persistent storage backend that saves results to the file system. Supports different serialization formats and directory structures.

Classes

FileSystemStorage

FileSystemStorage(base_path: str | Path, format: str = 'json', create_if_missing: bool = True, **config: Any)

Bases: StorageBackend

File system storage backend.

Stores results as files in a directory structure. Each result is saved with its metadata and sets in a dedicated subdirectory.

Initialize file system storage.

Parameters:

Name	Type	Description	Default
base_path	str | Path	Base directory for storage	required
format	str	File format for results ("json", "csv", etc.)	'json'
create_if_missing	bool	Create base directory if it doesn't exist	True
**config	Any	Additional configuration	{}

Source code in optiscope/storage/file_system.py

def __init__(
    self,
    base_path: str | Path,
    format: str = "json",
    create_if_missing: bool = True,
    **config: Any,
) -> None:
    """
    Initialize file system storage.

    Args:
        base_path: Base directory for storage
        format: File format for results ("json", "csv", etc.)
        create_if_missing: Create base directory if it doesn't exist
        **config: Additional configuration
    """
    self.base_path = Path(base_path)
    self.format = format
    self.create_if_missing = create_if_missing
    super().__init__(**config)

Functions

save_result

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

Save result to file system.

Source code in optiscope/storage/file_system.py

def save_result(
    self, key: str, result: OptimizationResult, metadata: dict[str, Any] | None = None
) -> None:
    """Save result to file system."""
    result_path = self._get_result_path(key)
    result_file = self._get_result_file(key)

    try:
        # Create directory
        result_path.mkdir(parents=True, exist_ok=True)

        # Save result data
        save_results(result, result_file, format_hint=self.format)

        # Create/update metadata
        now = datetime.now()
        meta = self._load_metadata(key) if self._get_metadata_file(key).exists() else {}

        if "created_at" not in meta:
            meta["created_at"] = now.isoformat()

        meta.update(
            {
                "key": key,
                "modified_at": now.isoformat(),
                "n_points": result.n_points,
                "n_sets": len(result.list_sets()),
                "format": self.format,
                "custom": metadata or {},
            }
        )

        self._save_metadata(key, meta)

        # Save result sets
        sets_dir = self._get_sets_dir(key)
        sets_dir.mkdir(exist_ok=True)

        for set_name, result_set in result.sets.items():
            self.save_set(key, set_name, result_set)

    except Exception as e:
        # Clean up on failure
        if result_path.exists():
            shutil.rmtree(result_path, ignore_errors=True)
        raise StorageOperationError(f"Failed to save result '{key}': {str(e)}") from e

load_result

load_result(key: str) -> OptimizationResult

Load result from file system.

Source code in optiscope/storage/file_system.py

def load_result(self, key: str) -> OptimizationResult:
    """Load result from file system."""
    result_file = self._get_result_file(key)

    if not result_file.exists():
        raise KeyError(f"Result '{key}' not found in storage")

    try:
        result = load_results(result_file, format_hint=self.format)

        # Load result sets
        sets_dir = self._get_sets_dir(key)
        if sets_dir.exists():
            for set_file in sets_dir.glob("*.json"):
                set_name = set_file.stem
                try:
                    result_set = self.load_set(key, set_name)
                    result._set_manager.add_set(result_set)
                except Exception as e:
                    # Log but don't fail on individual set load errors
                    print(f"Warning: Failed to load set '{set_name}': {e}")

        return result

    except Exception as e:
        raise StorageOperationError(f"Failed to load result '{key}': {str(e)}") from e

delete_result

delete_result(key: str) -> None

Delete result from file system.

Source code in optiscope/storage/file_system.py

def delete_result(self, key: str) -> None:
    """Delete result from file system."""
    result_path = self._get_result_path(key)

    if not result_path.exists():
        raise KeyError(f"Result '{key}' not found in storage")

    try:
        shutil.rmtree(result_path)
    except Exception as e:
        raise StorageOperationError(f"Failed to delete result '{key}': {str(e)}") from e

rename_result

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

Rename a result.

Source code in optiscope/storage/file_system.py

def rename_result(self, old_key: str, new_key: str) -> None:
    """Rename a result."""
    old_path = self._get_result_path(old_key)
    new_path = self._get_result_path(new_key)

    if not old_path.exists():
        raise KeyError(f"Result '{old_key}' not found in storage")

    if new_path.exists():
        raise StorageOperationError(f"Result '{new_key}' already exists")

    try:
        # Rename directory
        shutil.move(str(old_path), str(new_path))

        # Update metadata if it exists
        try:
            meta = self._load_metadata(new_key)
            meta["key"] = new_key
            meta["modified_at"] = datetime.now().isoformat()
            self._save_metadata(new_key, meta)
        except Exception as e:
            # If metadata update fails, we might want to warn but the rename happened
            print(f"Warning: Failed to update metadata for renamed result '{new_key}': {e}")

    except Exception as e:
        raise StorageOperationError(
            f"Failed to rename result '{old_key}' to '{new_key}': {str(e)}"
        ) from e

exists_result

exists_result(key: str) -> bool

Check if result exists.

Source code in optiscope/storage/file_system.py

def exists_result(self, key: str) -> bool:
    """Check if result exists."""
    return self._get_result_file(key).exists()

list_results

list_results(prefix: str | None = None) -> list[str]

List all result keys.

Source code in optiscope/storage/file_system.py

def list_results(self, prefix: str | None = None) -> list[str]:
    """List all result keys."""
    if not self.base_path.exists():
        return []

    keys = []
    for item in self.base_path.iterdir():
        if item.is_dir():
            result_file = item / f"result.{self.format}"
            if result_file.exists():
                key = item.name
                if prefix is None or key.startswith(prefix):
                    keys.append(key)

    return sorted(keys)

save_set

save_set(result_key: str, set_name: str, result_set: ResultSet) -> None

Save result set to file system.

Source code in optiscope/storage/file_system.py

def save_set(self, result_key: str, set_name: str, result_set: ResultSet) -> None:
    """Save result set to file system."""
    if not self.exists_result(result_key):
        raise KeyError(f"Result '{result_key}' not found in storage")

    set_file = self._get_set_file(result_key, set_name)
    set_file.parent.mkdir(parents=True, exist_ok=True)

    try:
        with open(set_file, "w") as f:
            json.dump(result_set.to_dict(), f, indent=2, default=str)

        # Update metadata
        meta = self._load_metadata(result_key)
        meta["modified_at"] = datetime.now().isoformat()
        sets_dir = self._get_sets_dir(result_key)
        meta["n_sets"] = len(list(sets_dir.glob("*.json"))) if sets_dir.exists() else 0
        self._save_metadata(result_key, meta)

    except Exception as e:
        raise StorageOperationError(f"Failed to save set '{set_name}': {str(e)}") from e

load_set

load_set(result_key: str, set_name: str) -> ResultSet

Load result set from file system.

Source code in optiscope/storage/file_system.py

def load_set(self, result_key: str, set_name: str) -> ResultSet:
    """Load result set from file system."""
    set_file = self._get_set_file(result_key, set_name)

    if not set_file.exists():
        raise KeyError(f"Set '{set_name}' not found for result '{result_key}'")

    try:
        with open(set_file) as f:
            data = json.load(f)
        return ResultSet.from_dict(data)
    except Exception as e:
        raise StorageOperationError(f"Failed to load set '{set_name}': {str(e)}") from e

delete_set

delete_set(result_key: str, set_name: str) -> None

Delete result set from file system.

Source code in optiscope/storage/file_system.py

def delete_set(self, result_key: str, set_name: str) -> None:
    """Delete result set from file system."""
    set_file = self._get_set_file(result_key, set_name)

    if not set_file.exists():
        raise KeyError(f"Set '{set_name}' not found for result '{result_key}'")

    try:
        set_file.unlink()

        # Update metadata
        meta = self._load_metadata(result_key)
        meta["modified_at"] = datetime.now().isoformat()
        sets_dir = self._get_sets_dir(result_key)
        meta["n_sets"] = len(list(sets_dir.glob("*.json"))) if sets_dir.exists() else 0
        self._save_metadata(result_key, meta)

    except Exception as e:
        raise StorageOperationError(f"Failed to delete set '{set_name}': {str(e)}") from e

list_sets

list_sets(result_key: str) -> list[str]

List all sets for a result.

Source code in optiscope/storage/file_system.py

def list_sets(self, result_key: str) -> list[str]:
    """List all sets for a result."""
    if not self.exists_result(result_key):
        raise KeyError(f"Result '{result_key}' not found in storage")

    sets_dir = self._get_sets_dir(result_key)
    if not sets_dir.exists():
        return []

    return sorted([f.stem for f in sets_dir.glob("*.json")])

get_result_metadata

get_result_metadata(key: str) -> dict[str, Any]

Get result metadata.

Source code in optiscope/storage/file_system.py

def get_result_metadata(self, key: str) -> dict[str, Any]:
    """Get result metadata."""
    if not self.exists_result(key):
        raise KeyError(f"Result '{key}' not found in storage")

    return self._load_metadata(key)

update_result_metadata

update_result_metadata(key: str, metadata: dict[str, Any]) -> None

Update result metadata.

Source code in optiscope/storage/file_system.py

def update_result_metadata(self, key: str, metadata: dict[str, Any]) -> None:
    """Update result metadata."""
    if not self.exists_result(key):
        raise KeyError(f"Result '{key}' not found in storage")

    meta = self._load_metadata(key)

    if "custom" not in meta:
        meta["custom"] = {}

    meta["custom"].update(metadata)
    meta["modified_at"] = datetime.now().isoformat()

    self._save_metadata(key, meta)

clear

clear(prefix: str | None = None) -> int

Clear stored results.

Source code in optiscope/storage/file_system.py

def clear(self, prefix: str | None = None) -> int:
    """Clear stored results."""
    keys = self.list_results(prefix)

    for key in keys:
        self.delete_result(key)

    return len(keys)

get_storage_info

get_storage_info() -> dict[str, Any]

Get storage information.

Source code in optiscope/storage/file_system.py

def get_storage_info(self) -> dict[str, Any]:
    """Get storage information."""
    # Calculate total size
    total_size = 0
    n_files = 0

    if self.base_path.exists():
        for item in self.base_path.rglob("*"):
            if item.is_file():
                total_size += item.stat().st_size
                n_files += 1

    size_mb = total_size / (1024 * 1024)

    return {
        "storage_type": "filesystem",
        "base_path": str(self.base_path.absolute()),
        "format": self.format,
        "n_results": len(self.list_results()),
        "total_size_mb": size_mb,
        "n_files": n_files,
        "exists": self.base_path.exists(),
    }

compact

compact() -> dict[str, Any]

Compact storage by removing orphaned files.

Returns:

Type	Description
dict[str, Any]	Dictionary with compaction statistics

Source code in optiscope/storage/file_system.py

def compact(self) -> dict[str, Any]:
    """
    Compact storage by removing orphaned files.

    Returns:
        Dictionary with compaction statistics
    """
    removed_files = 0
    freed_bytes = 0

    if not self.base_path.exists():
        return {"removed_files": 0, "freed_mb": 0.0}

    # Find directories without result files
    for item in self.base_path.iterdir():
        if item.is_dir():
            result_file = item / f"result.{self.format}"
            if not result_file.exists():
                # Orphaned directory
                size = sum(f.stat().st_size for f in item.rglob("*") if f.is_file())
                shutil.rmtree(item)
                removed_files += 1
                freed_bytes += size

    return {"removed_files": removed_files, "freed_mb": freed_bytes / (1024 * 1024)}

to_dict

to_dict() -> dict[str, Any]

Serialize the filesystem storage to a dictionary.

Source code in optiscope/storage/file_system.py

def to_dict(self) -> dict[str, Any]:
    """Serialize the filesystem storage to a dictionary."""
    return {
        "storage_type": self.storage_type.value,
        "base_path": str(self.base_path),
        "format": self.format,
        "config": self.config,
    }

from_dict classmethod

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

Deserialize filesystem storage from a dictionary.

Source code in optiscope/storage/file_system.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> FileSystemStorage:
    """Deserialize filesystem storage from a dictionary."""
    return cls(
        base_path=data["base_path"],
        format=data.get("format", "json"),
        **data.get("config", {}),
    )

Functions

Memory Storage

memory

In-memory storage backend for optimization results.

Fast storage backend that keeps all data in memory. Useful for interactive sessions and when persistence is not required.

Classes

MemoryStorage

MemoryStorage(max_memory_mb: float | None = None, eviction_policy: str = 'lru', **config: Any)

Bases: StorageBackend

In-memory storage backend.

Stores all results in memory using Python dictionaries. This is the fastest storage option but data is lost when the process ends. Supports optional memory limits and automatic eviction.

Initialize memory storage.

Parameters:

Name	Type	Description	Default
max_memory_mb	float | None	Maximum memory usage in MB (None = unlimited)	None
eviction_policy	str	Eviction policy when limit reached ("lru", "fifo")	'lru'
**config	Any	Additional configuration	{}

Source code in optiscope/storage/memory.py

def __init__(
    self, max_memory_mb: float | None = None, eviction_policy: str = "lru", **config: Any
) -> None:
    """
    Initialize memory storage.

    Args:
        max_memory_mb: Maximum memory usage in MB (None = unlimited)
        eviction_policy: Eviction policy when limit reached ("lru", "fifo")
        **config: Additional configuration
    """
    self.max_memory_mb = max_memory_mb
    self.eviction_policy = eviction_policy
    super().__init__(**config)

Functions

save_result

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

Save result to memory.

Source code in optiscope/storage/memory.py

def save_result(
    self, key: str, result: OptimizationResult, metadata: dict[str, Any] | None = None
) -> None:
    """Save result to memory."""
    # Check memory limit before saving
    self._check_memory_limit(key)

    # Deep copy to avoid external modifications
    self._results[key] = deepcopy(result)

    # Track creation
    if key not in self._creation_order:
        self._creation_order.append(key)

    # Initialize metadata
    now = datetime.now()
    self._metadata[key] = {
        "created_at": now,
        "modified_at": now,
        "n_points": result.n_points,
        "n_sets": len(result.list_sets()),
        "custom": metadata or {},
    }

    # Update access
    self._update_access(key)

    # Check memory after save
    self._check_memory_limit()

load_result

load_result(key: str) -> OptimizationResult

Load result from memory.

Source code in optiscope/storage/memory.py

def load_result(self, key: str) -> OptimizationResult:
    """Load result from memory."""
    if key not in self._results:
        raise KeyError(f"Result '{key}' not found in storage")

    self._update_access(key)
    # Return a deep copy to avoid external modifications
    return deepcopy(self._results[key])

delete_result

delete_result(key: str) -> None

Delete result from memory.

Source code in optiscope/storage/memory.py

def delete_result(self, key: str) -> None:
    """Delete result from memory."""
    if key not in self._results:
        raise KeyError(f"Result '{key}' not found in storage")

    del self._results[key]

    # Clean up sets
    if key in self._sets:
        del self._sets[key]

    # Clean up metadata
    if key in self._metadata:
        del self._metadata[key]

    # Clean up tracking
    if key in self._access_order:
        self._access_order.remove(key)
    if key in self._creation_order:
        self._creation_order.remove(key)

rename_result

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

Rename a result.

Source code in optiscope/storage/memory.py

def rename_result(self, old_key: str, new_key: str) -> None:
    """Rename a result."""
    if old_key not in self._results:
        raise KeyError(f"Result '{old_key}' not found in storage")

    if new_key in self._results:
        raise StorageOperationError(f"Result '{new_key}' already exists")

    # Move result
    self._results[new_key] = self._results.pop(old_key)

    # Move sets
    if old_key in self._sets:
        self._sets[new_key] = self._sets.pop(old_key)

    # Move metadata
    if old_key in self._metadata:
        self._metadata[new_key] = self._metadata.pop(old_key)
        self._metadata[new_key]["modified_at"] = datetime.now()

    # Update tracking
    if old_key in self._access_order:
        idx = self._access_order.index(old_key)
        self._access_order[idx] = new_key

    if old_key in self._creation_order:
        idx = self._creation_order.index(old_key)
        self._creation_order[idx] = new_key

exists_result

exists_result(key: str) -> bool

Check if result exists in memory.

Source code in optiscope/storage/memory.py

def exists_result(self, key: str) -> bool:
    """Check if result exists in memory."""
    return key in self._results

list_results

list_results(prefix: str | None = None) -> list[str]

List all result keys.

Source code in optiscope/storage/memory.py

def list_results(self, prefix: str | None = None) -> list[str]:
    """List all result keys."""
    keys = list(self._results.keys())
    if prefix:
        keys = [k for k in keys if k.startswith(prefix)]
    return sorted(keys)

save_set

save_set(result_key: str, set_name: str, result_set: ResultSet) -> None

Save result set.

Source code in optiscope/storage/memory.py

def save_set(self, result_key: str, set_name: str, result_set: ResultSet) -> None:
    """Save result set."""
    if result_key not in self._results:
        raise KeyError(f"Result '{result_key}' not found in storage")

    if result_key not in self._sets:
        self._sets[result_key] = {}

    # Deep copy the set
    self._sets[result_key][set_name] = deepcopy(result_set)

    # Update metadata
    if result_key in self._metadata:
        self._metadata[result_key]["modified_at"] = datetime.now()
        self._metadata[result_key]["n_sets"] = len(self._sets[result_key])

    self._update_access(result_key)

load_set

load_set(result_key: str, set_name: str) -> ResultSet

Load result set.

Source code in optiscope/storage/memory.py

def load_set(self, result_key: str, set_name: str) -> ResultSet:
    """Load result set."""
    if result_key not in self._results:
        raise KeyError(f"Result '{result_key}' not found in storage")

    if result_key not in self._sets or set_name not in self._sets[result_key]:
        raise KeyError(f"Set '{set_name}' not found for result '{result_key}'")

    self._update_access(result_key)
    return deepcopy(self._sets[result_key][set_name])

delete_set

delete_set(result_key: str, set_name: str) -> None

Delete result set.

Source code in optiscope/storage/memory.py

def delete_set(self, result_key: str, set_name: str) -> None:
    """Delete result set."""
    if result_key not in self._results:
        raise KeyError(f"Result '{result_key}' not found in storage")

    if result_key not in self._sets or set_name not in self._sets[result_key]:
        raise KeyError(f"Set '{set_name}' not found for result '{result_key}'")

    del self._sets[result_key][set_name]

    # Update metadata
    if result_key in self._metadata:
        self._metadata[result_key]["modified_at"] = datetime.now()
        self._metadata[result_key]["n_sets"] = len(self._sets[result_key])

list_sets

list_sets(result_key: str) -> list[str]

List all sets for a result.

Source code in optiscope/storage/memory.py

def list_sets(self, result_key: str) -> list[str]:
    """List all sets for a result."""
    if result_key not in self._results:
        raise KeyError(f"Result '{result_key}' not found in storage")

    if result_key not in self._sets:
        return []

    return sorted(self._sets[result_key].keys())

get_result_metadata

get_result_metadata(key: str) -> dict[str, Any]

Get result metadata.

Source code in optiscope/storage/memory.py

def get_result_metadata(self, key: str) -> dict[str, Any]:
    """Get result metadata."""
    if key not in self._results:
        raise KeyError(f"Result '{key}' not found in storage")

    meta = self._metadata.get(key, {})
    return {
        "key": key,
        "created_at": meta.get("created_at"),
        "modified_at": meta.get("modified_at"),
        "n_points": meta.get("n_points"),
        "n_sets": meta.get("n_sets"),
        "custom": meta.get("custom", {}),
    }

update_result_metadata

update_result_metadata(key: str, metadata: dict[str, Any]) -> None

Update result metadata.

Source code in optiscope/storage/memory.py

def update_result_metadata(self, key: str, metadata: dict[str, Any]) -> None:
    """Update result metadata."""
    if key not in self._results:
        raise KeyError(f"Result '{key}' not found in storage")

    if key not in self._metadata:
        self._metadata[key] = {}

    # Update custom metadata
    if "custom" not in self._metadata[key]:
        self._metadata[key]["custom"] = {}

    self._metadata[key]["custom"].update(metadata)
    self._metadata[key]["modified_at"] = datetime.now()

clear

clear(prefix: str | None = None) -> int

Clear stored results.

Source code in optiscope/storage/memory.py

def clear(self, prefix: str | None = None) -> int:
    """Clear stored results."""
    if prefix is None:
        count = len(self._results)
        self._results.clear()
        self._sets.clear()
        self._metadata.clear()
        self._access_order.clear()
        self._creation_order.clear()
        return count
    else:
        keys_to_delete = [k for k in self._results.keys() if k.startswith(prefix)]
        for key in keys_to_delete:
            self.delete_result(key)
        return len(keys_to_delete)

get_storage_info

get_storage_info() -> dict[str, Any]

Get storage information.

Source code in optiscope/storage/memory.py

def get_storage_info(self) -> dict[str, Any]:
    """Get storage information."""
    # Estimate memory usage
    total_size = sys.getsizeof(self._results)
    for result in self._results.values():
        total_size += sys.getsizeof(result)

    size_mb = total_size / (1024 * 1024)

    return {
        "storage_type": "memory",
        "n_results": len(self._results),
        "n_total_sets": sum(len(sets) for sets in self._sets.values()),
        "size_mb": size_mb,
        "max_memory_mb": self.max_memory_mb,
        "eviction_policy": self.eviction_policy,
        "usage_percent": (size_mb / self.max_memory_mb * 100) if self.max_memory_mb else None,
    }

get_result_reference

get_result_reference(key: str) -> OptimizationResult

Get direct reference to result (no copy).

WARNING: This bypasses the copy protection and allows direct modification of stored data. Use with caution.

Parameters:

Name	Type	Description	Default
key	str	Result key	required

Returns:

Type	Description
OptimizationResult	Direct reference to stored OptimizationResult

Source code in optiscope/storage/memory.py

def get_result_reference(self, key: str) -> OptimizationResult:
    """
    Get direct reference to result (no copy).

    WARNING: This bypasses the copy protection and allows direct
    modification of stored data. Use with caution.

    Args:
        key: Result key

    Returns:
        Direct reference to stored OptimizationResult
    """
    if key not in self._results:
        raise KeyError(f"Result '{key}' not found in storage")

    self._update_access(key)
    return self._results[key]

to_dict

to_dict() -> dict[str, Any]

Serialize the memory storage to a dictionary.

Source code in optiscope/storage/memory.py

def to_dict(self) -> dict[str, Any]:
    """Serialize the memory storage to a dictionary."""
    return {
        "storage_type": self.storage_type.value,
        "results": {key: result.to_dict() for key, result in self._results.items()},
        "config": self.config,
    }

from_dict classmethod

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

Deserialize memory storage from a dictionary.

Source code in optiscope/storage/memory.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> MemoryStorage:
    """Deserialize memory storage from a dictionary."""
    storage = cls(**data.get("config", {}))
    results = {
        key: OptimizationResult.from_dict(res_data)
        for key, res_data in data.get("results", {}).items()
    }
    storage._results = results
    # Re-initialize other attributes based on loaded results if necessary
    storage._access_order = list(results.keys())
    storage._creation_order = list(results.keys())
    return storage