base

base

AsyncResultsBackend

Bases: Generic[T], ABC

Abstract base class for async results storage backends.

Implementations should provide asynchronous storage and retrieval of Pydantic models.

clear abstractmethod async

clear() -> None

Asynchronously clear all stored results.

Source code in src/results_manager/async_backends/base.py

@abstractmethod
async def clear(self) -> None:
    """
    Asynchronously clear all stored results.
    """
    pass

delete abstractmethod async

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

Asynchronously 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/async_backends/base.py

@abstractmethod
async def delete(self, result_id: Union[str, List[str]]) -> bool:
    """
    Asynchronously 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 async

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

Asynchronously 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/async_backends/base.py

@abstractmethod
async def exists(self, result_id: Union[str, List[str]]) -> bool:
    """
    Asynchronously 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 async

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

Asynchronously 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/async_backends/base.py

@abstractmethod
async def get(self, 
             result_id: Union[str, List[str]], 
             model_class: Optional[Type[T]] = None,
             namespace: Optional[str] = None) -> T:
    """
    Asynchronously 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 async

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

Asynchronously 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/async_backends/base.py

@abstractmethod
async def list_ids(self, prefix: Union[str, List[str]] = None) -> List[str]:
    """
    Asynchronously 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 async

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

Asynchronously 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/async_backends/base.py

@abstractmethod
async 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:
    """
    Asynchronously 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