extensions

Extra tools for supporting Rago.

Modules:

• base –

  Extension base module.

• cache –

  Cache backends for Rago steps.

• logs –

  Logging configuration helpers for Rago pipelines.

Classes:

• Cache –

  Abstract cache backend used by pipeline steps.

• CacheFile –

  File-based cache backend implemented with joblib.

• Logs –

  Attach a mutable log dictionary to a step.

Cache

Abstract cache backend used by pipeline steps.

Methods:

• get_file_key –

  Normalize a cache key into a stable string representation.

• load –

  Load cached data for the given key.

• save –

  Persist data in the cache.

get_file_key

get_file_key(key: Any) -> str

Normalize a cache key into a stable string representation.

Source code in src/rago/extensions/cache.py

def get_file_key(self, key: Any) -> str:
    """Normalize a cache key into a stable string representation."""
    return sha256(str(key).encode('utf-8')).hexdigest()

load abstractmethod

load(key: Any) -> Any

Load cached data for the given key.

Source code in src/rago/extensions/cache.py

@abstractmethod
def load(self, key: Any) -> Any:
    """Load cached data for the given key."""

save abstractmethod

save(key: Any, data: Any) -> None

Persist data in the cache.

Source code in src/rago/extensions/cache.py

@abstractmethod
def save(self, key: Any, data: Any) -> None:
    """Persist data in the cache."""

CacheFile

CacheFile(target_dir: Path | str)

Bases: Cache

File-based cache backend implemented with joblib.

Methods:

• get_file_key –

  Normalize a cache key into a stable string representation.

• get_file_path –

  Return the file path for a given cache key.

• load –

  Load cached data if present.

• save –

  Persist cached data to disk.

Source code in src/rago/extensions/cache.py

def __init__(self, target_dir: Path | str) -> None:
    self.target_dir = Path(target_dir)
    self.target_dir.mkdir(parents=True, exist_ok=True)

get_file_key

get_file_key(key: Any) -> str

Normalize a cache key into a stable string representation.

Source code in src/rago/extensions/cache.py

def get_file_key(self, key: Any) -> str:
    """Normalize a cache key into a stable string representation."""
    return sha256(str(key).encode('utf-8')).hexdigest()

get_file_path

get_file_path(key: Any) -> Path

Return the file path for a given cache key.

Source code in src/rago/extensions/cache.py

def get_file_path(self, key: Any) -> Path:
    """Return the file path for a given cache key."""
    return self.target_dir / f'{self.get_file_key(key)}.pkl'

load

load(key: Any) -> Any

Load cached data if present.

Source code in src/rago/extensions/cache.py

def load(self, key: Any) -> Any:
    """Load cached data if present."""
    file_path = self.get_file_path(key)
    if not file_path.exists():
        return None
    return joblib.load(file_path)

save

save(key: Any, data: Any) -> None

Persist cached data to disk.

Source code in src/rago/extensions/cache.py

def save(self, key: Any, data: Any) -> None:
    """Persist cached data to disk."""
    file_path = self.get_file_path(key)
    joblib.dump(data, file_path)

Logs

Logs(target: dict[str, Any] | None = None)

Bases: ParametersBase

Attach a mutable log dictionary to a step.

Methods:

• apply –

  Merge additional configuration into this object.

• process –

  Return the input unchanged for configuration-only objects.

Attributes:

• params (dict[str, Any]) –

  Expose the underlying parameter mapping.

Source code in src/rago/extensions/logs.py

def __init__(self, target: dict[str, Any] | None = None) -> None:
    super().__init__(logs=target if target is not None else {})

params property

params: dict[str, Any]

Expose the underlying parameter mapping.

apply

apply(parameters: Any) -> None

Merge additional configuration into this object.

Source code in src/rago/base.py

def apply(self, parameters: Any) -> None:
    """Merge additional configuration into this object."""
    self.data.update(config_to_dict(parameters))

process

process(inp: Input) -> Output

Return the input unchanged for configuration-only objects.

Source code in src/rago/base.py

def process(self, inp: Input) -> Output:
    """Return the input unchanged for configuration-only objects."""
    return inp.to_output()