xtrace_sdk.integrations.local package

Submodules

xtrace_sdk.integrations.local.compute module

class xtrace_sdk.integrations.local.compute.LocalCompute(data_path: str, homomorphic_client: Type[HammingClientBase])

Bases: ComputeIntegrationBase

Local compute integration for testing purposes.

Parameters:

ComputeIntegrationBase (class) – interface for compute integrations

async compute_hamming_distances(query: int, pk: dict[slice('x', <class 'int'>, None), slice('n', <class 'int'>, None)] | dict[slice('g', <class 'int'>, None), slice('n', <class 'int'>, None)], **kwargs) → list[tuple[int, int]]

Compute the hamming distances between the query and the data in the local database.

Parameters:

• query (int) – Encrypted query to be used for comparison

• pk (PaillierPublicKey) – the public key used for encryption

Returns:

list of tuples with the id and the encrypted hamming distance

Return type:

List[Tuple[ChunkID,int]]

async get_chunk_by_id(chunk_ids: list[int]) → list[dict[slice('chunk_content', <class 'str'>, None), slice('meta_data', dict[str, str], None)]]

Get a chunk by its id.

Parameters:

chunk_ids (list[ChunkID]) – the ids of the chunks of interest

Returns:

the chunks corresponding to the given ids

Return type:

list[Chunk]

xtrace_sdk.integrations.local.storage module

class xtrace_sdk.integrations.local.storage.LocalStorage(path)

Bases: StorageIntegrationBase

load_db()

Load the database from the storage.

load_index()

Load the index from the storage.

async store_db(db)

Store the database in the storage.

async store_index(index)

Store the index in the storage.

Module contents