API Reference

Protocols

SearchEngine

Protocol for search engine implementations.

Defines the structural contract for all search backends. Implementations do not need to inherit from this class — any class providing these methods satisfies the protocol (structural subtyping).

Concrete backends may still inherit from SearchEngine to signal explicit conformance and benefit from IDE completions.

async def index(
    index_name: str,
    documents: list[dict[str, Any]]
) -> Result[bool, SearchError]

Index documents into the named index.

async def update(
    index_name: str,
    document_id: str,
    document: dict[str, Any]
) -> Result[bool, SearchError]

Update an existing document by ID.

async def delete(
    index_name: str,
    document_id: str
) -> Result[bool, SearchError]

Delete a document by ID.

async def search(
    index_name: str,
    query: str,
    filters: dict[str, Any] | None = None,
    limit: int = 20,
    offset: int = 0,
    sort: list[str] | None = None
) -> Result[SearchResponse, SearchError]

Execute a search query and return Ok(response) or Err(SearchError).

async def create_index(
    index_name: str,
    settings: dict[str, Any] | None = None
) -> Result[bool, SearchError]

Create a new index, optionally with settings.

async def delete_index(index_name: str) -> Result[bool, SearchError]

Delete an index and all its documents.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Return the operational health of the backend.

---

Classes

BackendType

Supported search backends.

---

FederatedResults

Combined results from searching across multiple indices.

def to_combined_list(include_index: bool = True) -> list[dict[str, Any]]

Convert federated results to a flat list.

Parameters

Parameter	Type	Description
`include_index`	bool	Whether to include the source index in each result.

Returns

Type	Description
list[dict[str, Any]]	Flat list of all results with optional index metadata.

---

FederatedSearchEngine

Federated search engine that searches across multiple indices.

This class wraps a primary search engine and allows searching across multiple indices simultaneously, combining and ranking results.

Example

# Create federated search across products and documents
federated = FederatedSearchEngine(
    engine=search_engine,
    indices=["products", "documents"],
)

# Search across all indices
results = await federated.search("laptop")

def __init__(
    engine: SearchEngineProtocol,
    indices: list[str] | None = None
) -> None

Initialize the federated search engine.

Parameters

Parameter	Type	Description
`engine`	SearchEngineProtocol	The underlying search engine to use.
`indices`	list[str] | None	Optional list of indices to search. Can be overridden per-query.

def set_indices(indices: list[str]) -> None

Set the default indices for federated searches.

Parameters

Parameter	Type	Description
`indices`	list[str]	List of index names to search.

async def search_across(
    query: str,
    indices: list[str] | None = None,
    filters: dict[str, Any] | None = None,
    sort: list[dict[str, str]] | None = None,
    limit_per_index: int | None = None,
    total_limit: int | None = None
) -> FederatedResults

Search across multiple indices.

Parameters

Parameter	Type	Description
`query`	str	Search query string.
`indices`	list[str] | None	List of indices to search (overrides default).
`filters`	dict[str, Any] | None	Optional filters to apply to each search.
`sort`	list[dict[str, str]] | None	Sort specifications.
`limit_per_index`	int | None	Maximum results per index.
`total_limit`	int | None	Maximum total results across all indices.

Returns

Type	Description
FederatedResults	FederatedResults containing results from each index.

async def search_with_fallback(
    query: str,
    primary_indices: list[str],
    fallback_indices: list[str] | None = None,
    min_results: int = 1,
    **kwargs: Any
) -> FederatedResults

Search with fallback to secondary indices.

Searches primary indices first, falls back to secondary indices if insufficient results are found.

Parameters

Parameter	Type	Description
`query`	str	Search query string.
`primary_indices`	list[str]	Primary indices to search first.
`fallback_indices`	list[str] | None	Secondary indices to search if needed.
`min_results`	int	Minimum results required before fallback. **kwargs: Additional search parameters.

Returns

Type	Description
FederatedResults	FederatedResults with combined results.

---

FilterCondition

A single field-operator-value predicate within a FilterSet.

Parameters

Parameter	Type	Description
`field`		The document field to filter on.
`operator`		The comparison operator to apply.
`value`		The filter value. ``None`` is valid for null-check operators.

---

FilterOperator

Operator used in an admin-facing filter condition.

---

FilterSet

A structured collection of filter conditions plus pagination/sorting.

Parameters

Parameter	Type	Description
`conditions`		Zero or more FilterCondition predicates that are ANDed together when translated to a SearchQuery.
`order_by`		Optional field name to sort by.
`order_dir`		Sort direction — ``"asc"`` (default) or ``"desc"``.
`page`		1-based page number (default ``1``).
`page_size`		Number of results per page (default ``25``).
`search_query`		Optional free-text query string passed as the ``q`` parameter of the resulting SearchQuery.

---

FilterSetTranslator

Translates a FilterSet into a SearchQuery.

All conditions in the FilterSet are composed with AND semantics — a document must satisfy every condition to appear in results.

Example

from lexigram.search.filterset import FilterCondition, FilterOperator, FilterSet
from lexigram.search.filterset import FilterSetTranslator

fs = FilterSet(
    conditions=(
        FilterCondition("status", FilterOperator.EQ, "active"),
        FilterCondition("score", FilterOperator.GTE, 80),
    ),
    order_by="name",
    order_dir="asc",
    page=2,
    page_size=10,
    search_query="python",
)

translator = FilterSetTranslator()
search_query = translator.translate(fs)

def translate(filter_set: FilterSet) -> SearchQuery

Translate filter_set to a SearchQuery.

Parameters

Parameter	Type	Description
`filter_set`	FilterSet	The admin-facing filter specification to convert.

Returns

Type	Description
SearchQuery	A SearchQuery ready to pass to a search engine.

Raises

Exception	Description
ValueError	If a FilterCondition carries an operator that cannot be mapped (should not happen with valid enum values).

---

InMemorySearchAnalyticsRecorder

In-memory search analytics recorder.

Stores search events in memory for testing and development. Production use should implement this protocol with a persistent store.

Example

recorder = InMemorySearchAnalyticsRecorder()
await recorder.record_search("python", None, 10, user_id="user123")
metrics = await recorder.get_search_metrics()

def __init__(max_events: int = 10000) -> None

Initialize the analytics recorder.

Parameters

Parameter	Type	Description
`max_events`	int	Maximum number of events to store in memory.

async def record_search(
    query: str,
    filters: dict[str, Any] | None,
    result_count: int,
    user_id: str | None = None,
    session_id: str | None = None
) -> None

Record a search query for analytics.

Parameters

Parameter	Type	Description
`query`	str	Search query string.
`filters`	dict[str, Any] | None	Applied filters.
`result_count`	int	Number of results returned.
`user_id`	str | None	User identifier (optional).
`session_id`	str | None	Session identifier (optional).

async def get_search_metrics(time_range: dict[str, Any] | None = None) -> dict[str, Any]

Get search analytics metrics.

Parameters

Parameter	Type	Description
`time_range`	dict[str, Any] | None	Time range for metrics (optional). Format: {"hours": 24} or {"days": 7} or {"start": "2024-01-01", "end": "2024-01-31"}

Returns

Type	Description
dict[str, Any]	Search metrics data including query counts, popular queries, etc.

async def get_zero_result_queries(time_range: dict[str, Any] | None = None) -> list[dict[str, Any]]

Get queries that returned zero results.

Parameters

Parameter	Type	Description
`time_range`	dict[str, Any] | None	Time range for metrics (optional).

Returns

Type	Description
list[dict[str, Any]]	List of zero-result queries with counts.

def clear() -> None

Clear all recorded events.

---

IndexConfig

Search index configuration.

---

IndexingCompletedEvent

Emitted when a batch indexing operation finishes.

Attributes: index_name: Name of the search index that was updated. documents_indexed: Number of documents successfully indexed.

---

NamedSearchConfig

Configuration for a single named search backend instance.

Used in SearchConfig.backends to declare multiple search backends that the framework registers as named DI bindings.

Example

backends:

• name: primary primary: true backend_type: meilisearch meilisearch: url: http://localhost:7700
• name: audit backend_type: postgres database: audit_db postgres: connection_string: postgresql://…

Parameters

Parameter	Type	Description
`name`		Unique backend identifier. Used as the Named() DI key.
`primary`		Whether this is the primary backend. Primary backends also receive the unnamed SearchEngine binding.
`backend_type`		The search backend to use for this entry.
`database`		Optional database name for DB-backed backends (postgres/mysql).
`meilisearch`		MeiliSearch-specific connection config.
`elasticsearch`		Elasticsearch-specific connection config.
`typesense`		Typesense-specific connection config.
`postgres`		PostgreSQL FTS-specific config.
`mysql`		MySQL FULLTEXT-specific config.
`sqlite`		SQLite FTS5-specific config.
`mongo`		MongoDB text search-specific config.

---

RAGSearchResult

Single search result.

Extends the canonical SearchIndexResult from contracts with DomainModel semantics.

---

SearchAnalyticsRecorder

In-memory search analytics recorder.

Stores search events in memory for testing and development. Production use should implement this protocol with a persistent store.

Example

recorder = InMemorySearchAnalyticsRecorder()
await recorder.record_search("python", None, 10, user_id="user123")
metrics = await recorder.get_search_metrics()

def __init__(max_events: int = 10000) -> None

Initialize the analytics recorder.

Parameters

Parameter	Type	Description
`max_events`	int	Maximum number of events to store in memory.

async def record_search(
    query: str,
    filters: dict[str, Any] | None,
    result_count: int,
    user_id: str | None = None,
    session_id: str | None = None
) -> None

Record a search query for analytics.

Parameters

Parameter	Type	Description
`query`	str	Search query string.
`filters`	dict[str, Any] | None	Applied filters.
`result_count`	int	Number of results returned.
`user_id`	str | None	User identifier (optional).
`session_id`	str | None	Session identifier (optional).

async def get_search_metrics(time_range: dict[str, Any] | None = None) -> dict[str, Any]

Get search analytics metrics.

Parameters

Parameter	Type	Description
`time_range`	dict[str, Any] | None	Time range for metrics (optional). Format: {"hours": 24} or {"days": 7} or {"start": "2024-01-01", "end": "2024-01-31"}

Returns

Type	Description
dict[str, Any]	Search metrics data including query counts, popular queries, etc.

async def get_zero_result_queries(time_range: dict[str, Any] | None = None) -> list[dict[str, Any]]

Get queries that returned zero results.

Parameters

Parameter	Type	Description
`time_range`	dict[str, Any] | None	Time range for metrics (optional).

Returns

Type	Description
list[dict[str, Any]]	List of zero-result queries with counts.

def clear() -> None

Clear all recorded events.

---

SearchConfig

Root configuration for Lexigram Search.

def coerce_backend_type(
    cls,
    v: object
) -> object

Accept plain strings as BackendType values.

def validate_for_backend() -> None

Validate that required config is present for the selected backend.

def from_named(
    cls,
    entry: NamedSearchConfig
) -> SearchConfig

Build a single-backend SearchConfig from a NamedSearchConfig entry.

Used internally by SearchProvider to create per-backend configs from a multi-backend declaration.

Parameters

Parameter	Type	Description
`entry`	NamedSearchConfig	The named backend entry to materialise.

Returns

Type	Description
SearchConfig	A SearchConfig configured for the single named backend.

---

SearchEntityRepository

Search-engine-backed repository implementing the platform RepositoryProtocol protocol.

Extends AbstractRepository with search-specific query capabilities, enabling search-backed repositories to satisfy the same RepositoryProtocol[T] protocol as database-backed repositories (Liskov Substitution Principle).

Subclasses must implement _to_document and _from_document to define entity ↔ search-document serialization, and may override _get_document_id if the entity ID is not stored in the "id" field of the document.

Example

class ProductRepository(SearchEntityRepository[Product]):
    def _to_document(self, entity: Product) -> dict[str, Any]:
        return {"id": entity.id, "name": entity.name, "price": entity.price}

    def _from_document(self, doc: dict[str, Any]) -> Product:
        return Product(**doc)

Parameters

Parameter	Type	Description
`engine`		The search engine backend to delegate operations to.
`index_name`		The name of the search index that stores entities of type ``T``.

def __init__(
    engine: SearchEngine,
    index_name: str
) -> None

async def find_by_spec(spec: SpecificationProtocol[T]) -> list[T]

Return all entities that satisfy the given specification.

Fetches all documents via a wildcard query and applies the specification in-memory. For large indices, prefer overriding this method to push filter logic into the search engine.

Parameters

Parameter	Type	Description
`spec`	SpecificationProtocol[T]	A ``SpecificationProtocol`` whose ``is_satisfied_by`` predicate is applied to each retrieved entity.

Returns

Type	Description
list[T]	All entities satisfying the specification.

async def save_many(entities: list[T]) -> list[T]

Index a batch of entities using a single bulk operation.

Prefer this over calling save in a loop when persisting multiple entities at once. Uses SearchEngine.index_many to issue a single bulk request, avoiding N separate index round-trips.

Parameters

Parameter	Type	Description
`entities`	list[T]	The list of entities to index.

Returns

Type	Description
list[T]	The same list of entities unchanged.

async def search(
    query: str,
    *,
    filters: dict[str, Any] | None = None,
    limit: int = 20,
    offset: int = 0,
    fields: list[str] | None = None
) -> Result[list[T], SearchError]

Execute a full-text search query against the index.

This is a search-specific extension on top of the base repository interface. Use this for user-facing search features rather than list() / find_by_spec().

Parameters

Parameter	Type	Description
`query`	str	The full-text query string.
`filters`	dict[str, Any] | None	Optional attribute equality filters to narrow results.
`limit`	int	Maximum number of results to return.
`offset`	int	Number of results to skip.
`fields`	list[str] | None	Optional list of fields to return; defaults to all fields.

Returns

Type	Description
Result[list[T], SearchError]	``Ok(list[T])`` on success, ``Err(SearchError)`` on backend failure.

async def find(entity_id: str) -> Result[T, NotFoundError]

Retrieve an entity by ID, returning a typed Result.

Provides an explicit Result-typed alternative to get() for callers that prefer structured error handling over None checks.

Parameters

Parameter	Type	Description
`entity_id`	str	The string ID of the entity to look up.

Returns

Type	Description
Result[T, NotFoundError]	``Ok(entity)`` if found, ``Err(NotFoundError)`` otherwise.

---

SearchExecutedEvent

Emitted when a search query is executed against the engine.

Attributes: index_name: Name of the index that was searched. query: The search query string. result_count: Number of results returned.

---

SearchIndexedHook

Payload fired when a document is added or updated in the search index.

Attributes: index_name: Name of the search index that was written to. document_id: Identifier of the document that was indexed.

---

SearchModule

Full-text and semantic search engine integration (Meilisearch, SQLite, Postgres).

Registers SearchEngineProtocol for constructor injection.

Call configure to configure the search backend, or stub for an isolated in-memory setup with no external service dependencies.

Usage

from lexigram.search.config import SearchConfig

@module(
    imports=[SearchModule.configure(SearchConfig(backend="meilisearch"))]
)
class AppModule(Module):
    pass

def configure(
    cls,
    config: SearchConfig | Any | None = None,
    enable_facets: bool = True
) -> DynamicModule

Create a SearchModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	SearchConfig | Any | None	SearchConfig or a pre-built SearchEngine instance. Passing ``None`` raises ``ValueError`` — a backend must be specified explicitly.
`enable_facets`	bool	Enable faceted search / filterable attributes on the backing index. Defaults to ``True``.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

Raises

Exception	Description
ValueError	If *config* is ``None``.

def stub(
    cls,
    config: SearchConfig | None = None
) -> DynamicModule

Create a SearchModule suitable for unit and integration testing.

Uses an in-memory (null) backend with no external service dependencies.

Parameters

Parameter	Type	Description
`config`	SearchConfig | None	Optional SearchConfig override. Uses an in-memory backend when ``None``.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

---

SearchProvider

Provider for search functionality.

Supports two registration modes:

Single-backend (default, existing behaviour): SearchConfig.backends is empty. Registers SearchProvider and SearchEngine as unnamed singletons, exactly as before.

Multi-backend (Named DI): SearchConfig.backends is non-empty. Each entry is registered as Annotated[SearchEngineProtocol, Named(entry.name)]. The primary backend (primary=True or the first entry) also receives the unnamed SearchEngineProtocol binding for backward compatibility. Database-backed backends (POSTGRES / MYSQL) use a NullBackend placeholder during register() and are resolved from the container in boot().

def __init__(
    backend: SearchEngine,
    config: SearchConfig | None = None
)

Initialize the provider with a configured backend.

def from_config(
    cls,
    config: SearchConfig,
    **context: Any
) -> SearchProvider

Create a SearchProvider from config.

Delegates to the existing configure() classmethod for backend selection.

async def register(container: ContainerRegistrarProtocol) -> None

Register search services with the container.

async def boot(container: ContainerResolverProtocol) -> None

Initialize the search backend(s) and verify connectivity.

In multi-backend mode all backends are health-checked in parallel via asyncio.gather. DB-backed backends are resolved from the container first, replacing the NullBackend placeholder registered during register().

In single-backend mode the existing sequential boot logic is preserved.

Raises

Exception	Description
RuntimeError	If any backend is unreachable at startup.

async def shutdown() -> None

Clean up search connections.

In multi-backend mode backends are closed in LIFO order (reversed registration order) so later-registered backends are torn down first.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Check health of search backend(s).

In multi-backend mode the overall status is the worst individual status (UNHEALTHY > DEGRADED > HEALTHY).

def configure(
    cls,
    config: SearchConfig
) -> SearchProvider

Create a SearchProvider configured from a SearchConfig object.

def with_meilisearch(
    cls,
    url: str = 'http://localhost:7700',
    api_key: str | None = None
) -> SearchProvider

Create provider with MeiliSearch backend.

def with_memory(cls) -> SearchProvider

Create provider with in-memory (null) backend.

---

SearchQuery

Search query parameters.

---

SearchQueryExecutedHook

Payload fired after a search query is executed.

Attributes: index_name: Name of the search index that was queried. query: The raw query string that was executed. result_count: Number of results returned.

---

SearchQueryValidator

Validator for search queries, filters, and other parameters.

def __init__(
    max_query_length: int = 1000,
    max_filter_length: int = 500
)

def validate_query(query: str) -> tuple[bool, str | None]

Validate a search query string.

def validate_filters(filters: Any) -> tuple[bool, str | None]

Validate search filters.

def validate_sort(sort: list[str] | None) -> tuple[bool, str | None]

Validate sort parameters.

def validate_index_name(name: str) -> tuple[bool, str | None]

Validate an index name.

def sanitize_query(query: str) -> str

Sanitize a search query.

def sanitize_filters(filters: Any) -> Any

Sanitize search filters.

---

SearchResponse

Search response with pagination.

---

SearchResult

Single search result.

Extends the canonical SearchIndexResult from contracts with DomainModel semantics.

---

SearchStrategy

Search strategies.

---

SearchableModel

Base class for models that can be searched.

def to_searchable_array() -> dict[str, Any]

Convert model to searchable data.

def get_searchable_key(cls) -> str

Get the searchable key (index name) for this model.

def get_searchable_settings(cls) -> dict[str, Any]

Get search settings for this model (defaults).

---

Suggestion

A single suggestion result.

---

SuggestionEngine

Suggestion engine providing autocomplete and typo-tolerant suggestions.

This engine provides:

• Prefix-based autocomplete suggestions
• Did-you-mean corrections for misspellings
• Popular/search frequency-based suggestions

Example

suggestion_engine = SuggestionEngine(
    search_engine=my_search_engine,
    index_name="products",
)

# Get autocomplete suggestions
suggestions = await suggestion_engine.suggest("lap")

# Get did-you-mean correction
correction = await suggestion_engine.did_you_mean("laptap")

def __init__(
    search_engine: SearchEngineProtocol,
    index_name: str = 'default',
    min_score: float = 0.5,
    max_suggestions: int = 10
) -> None

Initialize the suggestion engine.

Parameters

Parameter	Type	Description
`search_engine`	SearchEngineProtocol	The search engine to use for suggestions.
`index_name`	str	The index to search for suggestions.
`min_score`	float	Minimum score threshold for suggestions.
`max_suggestions`	int	Maximum number of suggestions to return.

async def suggest(
    prefix: str,
    field: str = 'name',
    filters: dict[str, Any] | None = None
) -> SuggestionResult

Get autocomplete suggestions for a prefix.

Parameters

Parameter	Type	Description
`prefix`	str	The prefix to get suggestions for.
`field`	str	The field to search for suggestions.
`filters`	dict[str, Any] | None	Optional filters to apply.

Returns

Type	Description
SuggestionResult	SuggestionResult with matching suggestions.

async def did_you_mean(
    query: str,
    max_variants: int = 3
) -> SuggestionResult

Get did-you-mean corrections for a query.

Performs simple typo detection and suggests corrections.

Parameters

Parameter	Type	Description
`query`	str	The misspelled query.
`max_variants`	int	Maximum number of correction variants to generate.

Returns

Type	Description
SuggestionResult	SuggestionResult with correction suggestions.

---

SuggestionResult

Result from a suggestion query.

---

Functions

sanitize_search_filters

async def sanitize_search_filters(filters: Any) -> Any

---

sanitize_search_query

async def sanitize_search_query(query: str) -> str

---

validate_index_name

async def validate_index_name(name: str) -> tuple[bool, str | None]

---

validate_search_filters

async def validate_search_filters(filters: Any) -> tuple[bool, str | None]

---

validate_search_query

async def validate_search_query(query: str) -> tuple[bool, str | None]

---

validate_search_sort

async def validate_search_sort(sort: list[str] | None) -> tuple[bool, str | None]

---

Exceptions

BackendError

Raised when search backend encounters an error.

---

ConfigurationError

Raised when search configuration is invalid.

---

QueryError

Raised when search query execution fails.

---

SearchError

Base exception for search operations.

---

SearchIndexError

Raised when search index operation fails.

---