file_backend

file_backend

AsyncFileBackend

AsyncFileBackend(
    base_dir: Union[str, Path],
    create_if_missing: bool = True,
    locks_dir: Optional[Union[str, Path]] = None,
)

Bases: AsyncResultsBackend[T]

Async wrapper for FileBackend.

Runs the synchronous FileBackend methods in a threadpool to avoid blocking the event loop.

Initialize the AsyncFileBackend.

Parameters:

Name	Type	Description	Default
base_dir	Union[str, Path]	Base directory to store results	required
create_if_missing	bool	Whether to create the directory if it doesn't exist	True
locks_dir	Optional[Union[str, Path]]	Directory to store lock files. If None, uses a system temp directory.	None

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

def __init__(self, base_dir: Union[str, Path], create_if_missing: bool = True, locks_dir: Optional[Union[str, Path]] = None):
    """
    Initialize the AsyncFileBackend.

    Args:
        base_dir: Base directory to store results
        create_if_missing: Whether to create the directory if it doesn't exist
        locks_dir: Directory to store lock files. If None, uses a system temp directory.
    """
    # Create the synchronous backend
    self._backend = FileBackend(base_dir, create_if_missing, locks_dir)

clear async

clear() -> None

Asynchronously clear all stored results.

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

async def clear(self) -> None:
    """
    Asynchronously clear all stored results.
    """
    await asyncio.to_thread(
        self._backend.clear
    )

delete 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/file_backend.py

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
    """
    return await asyncio.to_thread(
        self._backend.delete,
        result_id=result_id
    )

exists 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/file_backend.py

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
    """
    return await asyncio.to_thread(
        self._backend.exists,
        result_id=result_id
    )

get 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

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

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
    """
    return await asyncio.to_thread(
        self._backend.get,
        result_id=result_id,
        model_class=model_class,
        namespace=namespace
    )

list_ids 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/file_backend.py

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
    """
    return await asyncio.to_thread(
        self._backend.list_ids,
        prefix=prefix
    )

set 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

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

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
    """
    return await asyncio.to_thread(
        self._backend.set,
        result_id=result_id,
        data=data,
        behavior=behavior,
        namespace=namespace,
        strict_namespace=strict_namespace
    )