db

DBCache

DBCache(
    db_path: str = "cache.db",
    max_size: Optional[int] = None,
    time_to_live: Optional[int] = None,
    cleanup_interval: Optional[int] = None,
)

Bases: DBCacheBase

SQLite-based caching implementation.

Parameters:

• db_path (str, default: 'cache.db' ) –

  Path to the SQLite database file

Source code in src/appl/caching/db.py

def __init__(
    self,
    db_path: str = "cache.db",
    max_size: Optional[int] = None,
    time_to_live: Optional[int] = None,
    cleanup_interval: Optional[int] = None,
):
    """Initialize the database cache.

    Args:
        db_path: Path to the SQLite database file
    """
    self.db_path = Path(db_path)
    self.max_size = max_size or global_vars.configs.settings.caching.max_size
    self.time_to_live = (
        time_to_live or global_vars.configs.settings.caching.time_to_live
    )
    self.cleanup_interval = (
        cleanup_interval or global_vars.configs.settings.caching.cleanup_interval
    )
    self._init_db()
    self._init_cleanup_tracker()
    self._maybe_cleanup()

connect

connect() -> Connection

Create and return a database connection.

Returns:

• Connection –

  sqlite3.Connection: Database connection object

Source code in src/appl/caching/db.py

def connect(self) -> sqlite3.Connection:
    """Create and return a database connection.

    Returns:
        sqlite3.Connection: Database connection object
    """
    # Create parent directories if they don't exist
    os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
    return sqlite3.connect(self.db_path)

find

find(key: str) -> Optional[Any]

Find a value in the cache by key.

Parameters:

• key (str) –

  Cache key to look up

Returns:

• Optional[Any] –

  The cached value if found, None otherwise

Source code in src/appl/caching/db.py

def find(self, key: str) -> Optional[Any]:
    """Find a value in the cache by key.

    Args:
        key: Cache key to look up

    Returns:
        The cached value if found, None otherwise
    """
    self._maybe_cleanup()  # Only clean if needed

    uuid = encode_to_uuid_v5(key)

    with self.connect() as conn:
        cursor = conn.execute(
            "SELECT key, value FROM cache WHERE uuid = ?", (uuid,)
        )
        result = cursor.fetchone()

        if result is None:
            return None

        stored_key, value = result
        assert stored_key == key
        try:
            return json.loads(value)
        except json.JSONDecodeError:
            return value

insert

insert(
    key: str,
    value: Any,
    timestamp: Optional[datetime] = None,
) -> None

Insert a value to the cache.

Parameters:

• key (str) –

  Cache key

• value (Any) –

  Value to cache (will be JSON serialized)

Source code in src/appl/caching/db.py

def insert(
    self, key: str, value: Any, timestamp: Optional[datetime] = None
) -> None:
    """Insert a value to the cache.

    Args:
        key: Cache key
        value: Value to cache (will be JSON serialized)
    """
    self._maybe_cleanup()  # Only clean if needed

    uuid = encode_to_uuid_v5(key)

    serialized_value = json.dumps(value)
    if timestamp is None:
        timestamp = datetime.now()

    with self.connect() as conn:
        conn.execute(
            """
            INSERT OR REPLACE INTO cache (uuid, key, value, timestamp)
            VALUES (?, ?, ?, ?)
            """,
            (uuid, key, serialized_value, timestamp.isoformat()),
        )

add_to_cache

add_to_cache(
    args: Dict[str, Any],
    value: ModelResponse,
    cache: Optional[DBCacheBase] = None,
) -> None

Add a value to the LLM cache.

Source code in src/appl/caching/db.py

def add_to_cache(
    args: Dict[str, Any], value: ModelResponse, cache: Optional[DBCacheBase] = None
) -> None:
    """Add a value to the LLM cache."""
    cache = cache or global_vars.llm_cache
    if cache is None:
        logger.warning("No cache to add to")
        return
    if (
        args.get("temperature", 1.0) > 0.0
        and not global_vars.configs.settings.caching.allow_temp_greater_than_0
    ):
        return
    args_str = _serialize_args(args)
    value_dict = pydantic_to_dict(value)
    logger.info(f"Adding to cache, args: {args_str}, value: {value_dict}")
    cache.insert(args_str, value_dict)

find_in_cache

find_in_cache(
    args: Dict[str, Any],
    cache: Optional[DBCacheBase] = None,
) -> Optional[ModelResponse]

Find a value in the LLM cache by key.

Parameters:

• args (Dict[str, Any]) –

  The arguments of the completion.

• cache (Optional[DBCacheBase], default: None ) –

  The cache to search in. Defaults to the global LLM cache.

Returns:

• Optional[ModelResponse] –

  The completion result if found, otherwise None.

Source code in src/appl/caching/db.py

def find_in_cache(
    args: Dict[str, Any], cache: Optional[DBCacheBase] = None
) -> Optional[ModelResponse]:
    """Find a value in the LLM cache by key.

    Args:
        args: The arguments of the completion.
        cache: The cache to search in. Defaults to the global LLM cache.

    Returns:
        The completion result if found, otherwise None.
    """
    cache = cache or global_vars.llm_cache
    if cache is None:
        return None
    if (
        args.get("temperature", 1.0) > 0.0
        and not global_vars.configs.settings.caching.allow_temp_greater_than_0
    ):
        return None
    # only cache the completions with temperature == 0
    value = cache.find(_serialize_args(args))
    if value is None:
        return None
    return dict_to_pydantic(value, ModelResponse)