ahvn.klbase package

class ahvn.klbase.KLBase(storages=None, engines=None, name=None, *args, **kwargs)

Bases: ToolRegistry

Parameters:

• storages (List[BaseKLStore] | Dict[str, BaseKLStore] | None)

• engines (List[BaseKLEngine] | Dict[str, BaseKLEngine] | None)

• name (str | None)

__init__(storages=None, engines=None, name=None, *args, **kwargs)

Initialization.

Parameters:

• storages (Union[List[BaseKLStore], Dict[BaseKLStore]], optional) – A list or dictionary of storage backends. Defaults to None.

• engines (Union[List[BaseKLEngine], Dict[BaseKLEngine]], optional) – A list or dictionary of engine backends. Defaults to None.

• name (Optional[str], optional) – The name of this KLBase instance. If None, defaults to “default”.

• *args – Additional positional arguments.

• **kwargs – Additional keyword arguments.

add_storage(storage, name=None)

Add a storage backend.

Parameters:

• storage (BaseKLStore) – The storage backend to add.

• name (Optional[str], optional) – The name to register the storage under. If None, uses storage.name. Defaults to None

del_storage(name)

Delete a storage backend.

Parameters:

name (str) – The name of the storage backend to delete.

add_engine(engine, name=None, desync=False)

Add an engine backend.

Parameters:

• engine (BaseKLEngine) – The engine backend to add.

• name (Optional[str], optional) – The name to register the engine under. If None, uses engine.name. Defaults to None.

• desync (bool, optional) – Whether to desynchronize the engine upon addition. Defaults to False.

del_engine(name)

Delete an engine backend.

Parameters:

name (str) – The name of the engine backend to delete.

desync_engine(name)

Desynchronize an engine backend.

Parameters:

name (str) – The name of the engine backend to desynchronize.

resync_engine(name)

Synchronize an engine backend.

Parameters:

name (str) – The name of the engine backend to synchronize.

upsert(kl, storages=None, engines=None, **kwargs)

Upsert a KL into all storages and engines.

Parameters:

• kl (BaseUKF) – The KL to upsert.

• storages (List[str], optional) – The list of storage names to upsert into. If None and engines is None/empty, upserts into all storages. Defaults to None. If None and engines is not None/empty, upserts into no storages.

• engines (List[str], optional) – The list of engine names to upsert into. If None and storages is None/empty, upserts into all engines. Defaults to None. If None and storages is not None/empty, upserts into no engines.

• **kwargs – Additional keyword arguments.

insert(kl, storages=None, engines=None, **kwargs)

Insert a KL into all storages and engines.

Parameters:

• kl (BaseUKF) – The KL to insert.

• storages (List[str], optional) – The list of storage names to insert into. If None and engines is None/empty, inserts into all storages. Defaults to None. If None and engines is not None/empty, inserts into no storages.

• engines (List[str], optional) – The list of engine names to insert into. If None and storages is None/empty, inserts into all engines. Defaults to None. If None and storages is not None/empty, inserts into no engines.

• **kwargs – Additional keyword arguments.

batch_upsert(kls, storages=None, engines=None, progress=None, **kwargs)

Batch upsert KLs into all storages and engines.

Parameters:

• kls (List[BaseUKF]) – The list of KLs to upsert.

• storages (List[str], optional) – The list of storage names to upsert into. If None and engines is None/empty, batch upserts into all storages. Defaults to None. If None and engines is not None/empty, batch upserts into no storages.

• engines (List[str], optional) – The list of engine names to upsert into. If None, batch upserts into all engines. Defaults to None. If None and storages is None/empty, batch upserts into all engines. If None and storages is not None/empty, batch upserts into no engines.

• **kwargs – Additional keyword arguments.

• progress (Type[Progress])

batch_insert(kls, storages=None, engines=None, progress=None, **kwargs)

Batch insert KLs into all storages and engines.

Parameters:

• kls (List[BaseUKF]) – The list of KLs to insert.

• storages (List[str], optional) – The list of storage names to insert into. If None and engines is None/empty, batch inserts into all storages. Defaults to None. If None and engines is not None/empty, batch inserts into no storages.

• engines (List[str], optional) – The list of engine names to insert into. If None and storages is None/empty, batch inserts into all engines. Defaults to None. If None and storages is not None/empty, batch inserts into no engines.

• **kwargs – Additional keyword arguments.

• progress (Type[Progress])

remove(key, storages=None, engines=None, **kwargs)

Remove a KL from all storages and engines.

Parameters:

• key (Union[int, str, BaseUKF]) – The key or BaseUKF instance of the KL to remove.

• storages (List[str], optional) – The list of storage names to remove from. If None and engines is None/empty, removes from all storages. Defaults to None. If None and engines is not None/empty, removes from no storages.

• engines (List[str], optional) – The list of engine names to remove from. If None and storages is None/empty, removes from all engines. Defaults to None. If None and storages is not None/empty, removes from no engines.

• **kwargs – Additional keyword arguments.

batch_remove(keys, storages=None, engines=None, progress=None, **kwargs)

Batch remove KLs from all storages and engines.

Parameters:

• keys (List[Union[int, str, BaseUKF]]) – The list of keys or BaseUKF instances of the KLs to remove.

• storages (List[str], optional) – The list of storage names to remove from. If None and engines is None/empty, batch removes from all storages. Defaults to None. If None and engines is not None/empty, batch removes from no storages.

• engines (List[str], optional) – The list of engine names to remove from. If None and storages is None/empty, batch removes from all engines. Defaults to None. If None and storages is not None/empty, batch removes from no engines.

• **kwargs – Additional keyword arguments.

• progress (Type[Progress])

clear(storages=None, engines=None)

Clear all KLs from all storages and engines.

Parameters:

• storages (List[str], optional) – The list of storage names to clear. If None and engines is None/empty, clears all storages. Defaults to None. If None and engines is not None/empty, clears no storages.

• engines (List[str], optional) – The list of engine names to clear. If None and storages is None/empty, clears all engines. Defaults to None. If None and storages is not None/empty, clears no engines.

set_default_engine(name)

Set the default engine for searches.

Parameters:

name (str) – The name of the engine to set as default.

search(engine=None, *args, **kwargs)

Search for KLs using a specified engine.

Parameters:

• engine (Optional[str]) – The name of the engine to use for searching. If None, uses the default engine. If no default engine is set, raises a ValueError. Defaults to None.

• *args – Additional positional arguments for the engine’s search method.

• **kwargs – Additional keyword arguments for the engine’s search method.

Returns:

A list of dictionary, each being one search result.

As a recommended convention, all engine searches should support at least these keys:

• ”id” (int): the identifier of the KL.

• ”kl” (BaseUKF): the KL instance, if the engine is recoverable.

Other keys can be added as needed, which are usually search-specific (e.g., vector search score).

Return type:

Iterable[Dict[str, Any]]

list_search()

List all available engine search methods.

Returns:

A list of search method names, each a Tuple:

The first string is the engine to call. The second (optional) string is the engine’s search mode.

Return type:

List[Tuple[str, Optional[str]]]

sync(include_desynced=False, progress=None, **kwargs)

Sync all engines (one-time operation).

Parameters:

• include_desynced (bool) – Whether to sync desynced engines as well. Default is False.

• progress (Type[Progress]) – Progress class for reporting. None for silent, TqdmProgress for terminal.

• **kwargs – Additional keyword arguments for the engine’s sync method.

sync_desynced(progress=None, **kwargs)

Sync all desynced engines (one-time operation).

Parameters:

• progress (Type[Progress]) – Progress class for reporting. None for silent, TqdmProgress for terminal.

• **kwargs – Additional keyword arguments for the engine’s sync method.

flush()

Flush all storages and engines.

close()

Close all storages and engines.

Submodules

• ahvn.klbase.base module
  • KLBase
    • KLBase.__init__()
    • KLBase.add_storage()
    • KLBase.del_storage()
    • KLBase.add_engine()
    • KLBase.del_engine()
    • KLBase.desync_engine()
    • KLBase.resync_engine()
    • KLBase.upsert()
    • KLBase.insert()
    • KLBase.batch_upsert()
    • KLBase.batch_insert()
    • KLBase.remove()
    • KLBase.batch_remove()
    • KLBase.clear()
    • KLBase.set_default_engine()
    • KLBase.search()
    • KLBase.list_search()
    • KLBase.sync()
    • KLBase.sync_desynced()
    • KLBase.flush()
    • KLBase.close()