Fulltext

In-memory implementation of fulltext search and CRUD capability.

This module provides an in-memory implementation of the FulltextCapability protocol using dictionary-based storage optimized for development and testing scenarios.

Authors

Kadek Denaya (kadek.d.r.diana@gdplabs.id)

References

NONE

InMemoryFulltextCapability(store=None)

In-memory implementation of FulltextCapability protocol.

This class provides document CRUD operations and flexible querying using pure Python data structures optimized for development and testing.

Attributes:

Name	Type	Description
store	dict[str, Chunk]	Dictionary storing Chunk objects with their IDs as keys.

Initialize the in-memory fulltext capability.

Parameters:

Name	Type	Description	Default
store	dict[str, Any] | None	Dictionary storing Chunk objects with their IDs as keys. Defaults to None.	None

clear() async

Clear all records from the datastore.

create(data) async

Create new records in the datastore.

Example

Create a new chunk.

await fulltext_capability.create(Chunk(content="Test chunk", metadata={"category": "test"}))

Parameters:

Name	Type	Description	Default
data	Chunk | list[Chunk]	Data to create (single item or collection).	required

Raises:

Type	Description
ValueError	If data structure is invalid.

delete(filters=None) async

Delete records from the datastore.

Parameters:

Name	Type	Description	Default
filters	QueryFilter | None	Filters to select records to delete. Defaults to None, in which	None

retrieve(filters=None, options=None) async

Read records from the datastore with optional filtering.

Parameters:

Name	Type	Description	Default
filters	QueryFilter | None	Query filters to apply. Defaults to None.	None
options	QueryOptions | None	Query options (sorting, pagination, etc.). Defaults to None.	None

Returns:

Type	Description
list[Chunk]	list[Chunk]: Query results.

update(update_values, filters=None) async

Update existing records in the datastore.

Example

Update certain metadata of a chunk with specific filters.

await fulltext_capability.update(
    update_values={"metadata": {"status": "published"}},
    filters=QueryFilter(conditions={"metadata.status": "draft"}),
)

Parameters:

Name	Type	Description	Default
update_values	dict[str, Any]	Values to update.	required
filters	QueryFilter | None	Filters to select records to update. Defaults to None.	None