Skip to content

base

base

ResultsBackend

Bases: Generic[T], ABC

Abstract base class for results storage backends.

Implementations should provide storage and retrieval of Pydantic models based on unique IDs.

clear abstractmethod

clear() -> None

Clear all stored results.

Source code in src/results_manager/backends/base.py
@abstractmethod
def clear(self) -> None:
    """
    Clear all stored results.
    """
    pass

delete abstractmethod

delete(result_id: Union[str, List[str]]) -> bool

Delete a result by ID.

Parameters:

Name Type Description Default
result_id Union[str, List[str]]

Unique identifier or hierarchical path for the result

required

Returns:

Type Description
bool

True if deleted, False if not found

Source code in src/results_manager/backends/base.py
@abstractmethod
def delete(self, result_id: Union[str, List[str]]) -> bool:
    """
    Delete a result by ID.

    Args:
        result_id: Unique identifier or hierarchical path for the result

    Returns:
        True if deleted, False if not found
    """
    pass

exists abstractmethod

exists(result_id: Union[str, List[str]]) -> bool

Check if a result exists for the given ID.

Parameters:

Name Type Description Default
result_id Union[str, List[str]]

Unique identifier or hierarchical path for the result

required

Returns:

Type Description
bool

True if the result exists, False otherwise

Source code in src/results_manager/backends/base.py
@abstractmethod
def exists(self, result_id: Union[str, List[str]]) -> bool:
    """
    Check if a result exists for the given ID.

    Args:
        result_id: Unique identifier or hierarchical path for the result

    Returns:
        True if the result exists, False otherwise
    """
    pass

get abstractmethod

get(
    result_id: Union[str, List[str]],
    model_class: Optional[Type[T]] = None,
    namespace: Optional[str] = None,
) -> T

Retrieve a result by ID.

Parameters:

Name Type Description Default
result_id Union[str, List[str]]

Unique identifier or hierarchical path for the result

required
model_class Optional[Type[T]]

Optional model class to validate against

None
namespace Optional[str]

Optional namespace to look in

None

Returns:

Type Description
T

Pydantic model instance

Raises:

Type Description
FileNotFoundError

If the result doesn't exist

ValueError

If the model type is not registered

Source code in src/results_manager/backends/base.py
@abstractmethod
def get(self, 
        result_id: Union[str, List[str]], 
        model_class: Optional[Type[T]] = None,
        namespace: Optional[str] = None) -> T:
    """
    Retrieve a result by ID.

    Args:
        result_id: Unique identifier or hierarchical path for the result
        model_class: Optional model class to validate against
        namespace: Optional namespace to look in

    Returns:
        Pydantic model instance

    Raises:
        FileNotFoundError: If the result doesn't exist
        ValueError: If the model type is not registered
    """
    pass

list_ids abstractmethod

list_ids(prefix: Union[str, List[str]] = None) -> List[str]

List all result IDs, optionally filtered by a prefix.

Parameters:

Name Type Description Default
prefix Union[str, List[str]]

Optional prefix path to filter results

None

Returns:

Type Description
List[str]

List of result IDs

Source code in src/results_manager/backends/base.py
@abstractmethod
def list_ids(self, prefix: Union[str, List[str]] = None) -> List[str]:
    """
    List all result IDs, optionally filtered by a prefix.

    Args:
        prefix: Optional prefix path to filter results

    Returns:
        List of result IDs
    """
    pass

set abstractmethod

set(
    result_id: Union[str, List[str]],
    data: BaseModel,
    behavior: SetBehavior = RAISE_IF_EXISTS,
    namespace: Optional[str] = None,
    strict_namespace: bool = False,
) -> bool

Store a result with the given ID.

Parameters:

Name Type Description Default
result_id Union[str, List[str]]

Unique identifier or hierarchical path for the result

required
data BaseModel

Pydantic model instance to store

required
behavior SetBehavior

How to handle existing data with the same ID

RAISE_IF_EXISTS
namespace Optional[str]

Optional namespace to store the model in

None
strict_namespace bool

If True, raises an error if model is in multiple namespaces

False

Returns:

Type Description
bool

True if data was written, False if skipped

Raises:

Type Description
FileExistsError

If data already exists (for RAISE_IF_EXISTS) or if different data exists (for RAISE_IF_DIFFERENT)

Source code in src/results_manager/backends/base.py
@abstractmethod
def set(self, 
        result_id: Union[str, List[str]], 
        data: BaseModel, 
        behavior: SetBehavior = SetBehavior.RAISE_IF_EXISTS,
        namespace: Optional[str] = None,
        strict_namespace: bool = False) -> bool:
    """
    Store a result with the given ID.

    Args:
        result_id: Unique identifier or hierarchical path for the result
        data: Pydantic model instance to store
        behavior: How to handle existing data with the same ID
        namespace: Optional namespace to store the model in
        strict_namespace: If True, raises an error if model is in multiple namespaces

    Returns:
        True if data was written, False if skipped

    Raises:
        FileExistsError: If data already exists (for RAISE_IF_EXISTS) or
                         if different data exists (for RAISE_IF_DIFFERENT)
    """
    pass

SetBehavior

Bases: Enum

Defines behavior when setting data for an ID that already exists.