Skip to content
GitHub

API Reference

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.

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

Index documents into the named index.

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

Update an existing document by ID.

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

Delete a document by ID.

search
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).

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

Create a new index, optionally with settings.

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

Delete an index and all its documents.

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

Check if an index exists.

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

Return the operational health of the backend.


Supported search backends.

Combined results from searching across multiple indices.
to_combined_list
def to_combined_list(include_index: bool = True) -> list[dict[str, Any]]

Convert federated results to a flat list.

Parameters
ParameterTypeDescription
`include_index`boolWhether to include the source index in each result.
Returns
TypeDescription
list[dict[str, Any]]Flat list of all results with optional index metadata.

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")
# Create federated search across products and documents
federated = FederatedSearchEngine(
engine=search_engine,
indices=["products", "documents"],
)
# Search across all indices
results = await federated.search("laptop")
__init__
def __init__(
    engine: SearchEngineProtocol,
    indices: list[str] | None = None
) -> None

Initialize the federated search engine.

Parameters
ParameterTypeDescription
`engine`SearchEngineProtocolThe underlying search engine to use.
`indices`list[str] | NoneOptional list of indices to search. Can be overridden per-query.
set_indices
def set_indices(indices: list[str]) -> None

Set the default indices for federated searches.

Parameters
ParameterTypeDescription
`indices`list[str]List of index names to search.
search_across
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
ParameterTypeDescription
`query`strSearch query string.
`indices`list[str] | NoneList of indices to search (overrides default).
`filters`dict[str, Any] | NoneOptional filters to apply to each search.
`sort`list[dict[str, str]] | NoneSort specifications.
`limit_per_index`int | NoneMaximum results per index.
`total_limit`int | NoneMaximum total results across all indices.
Returns
TypeDescription
FederatedResultsFederatedResults containing results from each index.
search_with_fallback
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
ParameterTypeDescription
`query`strSearch query string.
`primary_indices`list[str]Primary indices to search first.
`fallback_indices`list[str] | NoneSecondary indices to search if needed.
`min_results`intMinimum results required before fallback. **kwargs: Additional search parameters.
Returns
TypeDescription
FederatedResultsFederatedResults with combined results.

A single field-operator-value predicate within a FilterSet.
Parameters
ParameterTypeDescription
`field`The document field to filter on.
`operator`The comparison operator to apply.
`value`The filter value. ``None`` is valid for null-check operators.

Operator used in an admin-facing filter condition.

A structured collection of filter conditions plus pagination/sorting.
Parameters
ParameterTypeDescription
`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.

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)
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)
translate
def translate(filter_set: FilterSet) -> SearchQuery

Translate filter_set to a SearchQuery.

Parameters
ParameterTypeDescription
`filter_set`FilterSetThe admin-facing filter specification to convert.
Returns
TypeDescription
SearchQueryA SearchQuery ready to pass to a search engine.
Raises
ExceptionDescription
ValueErrorIf a FilterCondition carries an operator that cannot be mapped (should not happen with valid enum values).

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()
recorder = InMemorySearchAnalyticsRecorder()
await recorder.record_search("python", None, 10, user_id="user123")
metrics = await recorder.get_search_metrics()
__init__
def __init__(max_events: int = 10000) -> None

Initialize the analytics recorder.

Parameters
ParameterTypeDescription
`max_events`intMaximum number of events to store in memory.
record_search
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
ParameterTypeDescription
`query`strSearch query string.
`filters`dict[str, Any] | NoneApplied filters.
`result_count`intNumber of results returned.
`user_id`str | NoneUser identifier (optional).
`session_id`str | NoneSession identifier (optional).
get_search_metrics
async def get_search_metrics(time_range: dict[str, Any] | None = None) -> dict[str, Any]

Get search analytics metrics.

Parameters
ParameterTypeDescription
`time_range`dict[str, Any] | NoneTime range for metrics (optional). Format: {"hours": 24} or {"days": 7} or {"start": "2024-01-01", "end": "2024-01-31"}
Returns
TypeDescription
dict[str, Any]Search metrics data including query counts, popular queries, etc.
get_zero_result_queries
async def get_zero_result_queries(time_range: dict[str, Any] | None = None) -> list[dict[str, Any]]

Get queries that returned zero results.

Parameters
ParameterTypeDescription
`time_range`dict[str, Any] | NoneTime range for metrics (optional).
Returns
TypeDescription
list[dict[str, Any]]List of zero-result queries with counts.
clear
def clear() -> None

Clear all recorded events.


Search index configuration.

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.


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
ParameterTypeDescription
`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.

Single search result.

Extends the canonical SearchIndexResult from contracts with DomainModel semantics.


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()
recorder = InMemorySearchAnalyticsRecorder()
await recorder.record_search("python", None, 10, user_id="user123")
metrics = await recorder.get_search_metrics()
__init__
def __init__(max_events: int = 10000) -> None

Initialize the analytics recorder.

Parameters
ParameterTypeDescription
`max_events`intMaximum number of events to store in memory.
record_search
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
ParameterTypeDescription
`query`strSearch query string.
`filters`dict[str, Any] | NoneApplied filters.
`result_count`intNumber of results returned.
`user_id`str | NoneUser identifier (optional).
`session_id`str | NoneSession identifier (optional).
get_search_metrics
async def get_search_metrics(time_range: dict[str, Any] | None = None) -> dict[str, Any]

Get search analytics metrics.

Parameters
ParameterTypeDescription
`time_range`dict[str, Any] | NoneTime range for metrics (optional). Format: {"hours": 24} or {"days": 7} or {"start": "2024-01-01", "end": "2024-01-31"}
Returns
TypeDescription
dict[str, Any]Search metrics data including query counts, popular queries, etc.
get_zero_result_queries
async def get_zero_result_queries(time_range: dict[str, Any] | None = None) -> list[dict[str, Any]]

Get queries that returned zero results.

Parameters
ParameterTypeDescription
`time_range`dict[str, Any] | NoneTime range for metrics (optional).
Returns
TypeDescription
list[dict[str, Any]]List of zero-result queries with counts.
clear
def clear() -> None

Clear all recorded events.


Root configuration for Lexigram Search.
coerce_backend_type
def coerce_backend_type(
    cls,
    v: object
) -> object

Accept plain strings as BackendType values.

validate_for_backend
def validate_for_backend() -> None

Validate that required config is present for the selected backend.

from_named
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
ParameterTypeDescription
`entry`NamedSearchConfigThe named backend entry to materialise.
Returns
TypeDescription
SearchConfigA SearchConfig configured for the single named backend.

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)
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
ParameterTypeDescription
`engine`The search engine backend to delegate operations to.
`index_name`The name of the search index that stores entities of type ``T``.
__init__
def __init__(
    engine: SearchEngine,
    index_name: str
) -> None
find_by_spec
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
ParameterTypeDescription
`spec`SpecificationProtocol[T]A ``SpecificationProtocol`` whose ``is_satisfied_by`` predicate is applied to each retrieved entity.
Returns
TypeDescription
list[T]All entities satisfying the specification.
save_many
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
ParameterTypeDescription
`entities`list[T]The list of entities to index.
Returns
TypeDescription
list[T]The same list of entities unchanged.
search
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
ParameterTypeDescription
`query`strThe full-text query string.
`filters`dict[str, Any] | NoneOptional attribute equality filters to narrow results.
`limit`intMaximum number of results to return.
`offset`intNumber of results to skip.
`fields`list[str] | NoneOptional list of fields to return; defaults to all fields.
Returns
TypeDescription
Result[list[T], SearchError]``Ok(list[T])`` on success, ``Err(SearchError)`` on backend failure.
find
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
ParameterTypeDescription
`entity_id`strThe string ID of the entity to look up.
Returns
TypeDescription
Result[T, NotFoundError]``Ok(entity)`` if found, ``Err(NotFoundError)`` otherwise.

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.


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.


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
from lexigram.search.config import SearchConfig
@module(
imports=[SearchModule.configure(SearchConfig(backend="meilisearch"))]
)
class AppModule(Module):
pass
configure
def configure(
    cls,
    config: SearchConfig | Any | None = None,
    enable_facets: bool = True
) -> DynamicModule

Create a SearchModule with explicit configuration.

Parameters
ParameterTypeDescription
`config`SearchConfig | Any | NoneSearchConfig or a pre-built SearchEngine instance. Passing ``None`` raises ``ValueError`` — a backend must be specified explicitly.
`enable_facets`boolEnable faceted search / filterable attributes on the backing index. Defaults to ``True``.
Returns
TypeDescription
DynamicModuleA DynamicModule descriptor.
Raises
ExceptionDescription
ValueErrorIf *config* is ``None``.
stub
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
ParameterTypeDescription
`config`SearchConfig | NoneOptional SearchConfig override. Uses an in-memory backend when ``None``.
Returns
TypeDescription
DynamicModuleA DynamicModule descriptor.

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().

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

Initialize the provider with a configured backend.

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

Create a SearchProvider from config.

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

register
async def register(container: ContainerRegistrarProtocol) -> None

Register search services with the container.

boot
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
ExceptionDescription
RuntimeErrorIf any backend is unreachable at startup.
shutdown
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.

health_check
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).

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

Create a SearchProvider configured from a SearchConfig object.

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

Create provider with MeiliSearch backend.

with_memory
def with_memory(cls) -> SearchProvider

Create provider with in-memory (null) backend.


Search query parameters.

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.


Validator for search queries, filters, and other parameters.
__init__
def __init__(
    max_query_length: int = 1000,
    max_filter_length: int = 500
)
validate_query
def validate_query(query: str) -> tuple[bool, str | None]

Validate a search query string.

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

Validate search filters.

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

Validate sort parameters.

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

Validate an index name.

sanitize_query
def sanitize_query(query: str) -> str

Sanitize a search query.

sanitize_filters
def sanitize_filters(filters: Any) -> Any

Sanitize search filters.


Search response with pagination.

Single search result.

Extends the canonical SearchIndexResult from contracts with DomainModel semantics.


Search strategies.

Base class for models that can be searched.
to_searchable_array
def to_searchable_array() -> dict[str, Any]

Convert model to searchable data.

get_searchable_key
def get_searchable_key(cls) -> str

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

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

Get search settings for this model (defaults).


A single suggestion result.

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")
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")
__init__
def __init__(
    search_engine: SearchEngineProtocol,
    index_name: str = 'default',
    min_score: float = 0.5,
    max_suggestions: int = 10
) -> None

Initialize the suggestion engine.

Parameters
ParameterTypeDescription
`search_engine`SearchEngineProtocolThe search engine to use for suggestions.
`index_name`strThe index to search for suggestions.
`min_score`floatMinimum score threshold for suggestions.
`max_suggestions`intMaximum number of suggestions to return.
suggest
async def suggest(
    prefix: str,
    field: str = 'name',
    filters: dict[str, Any] | None = None
) -> SuggestionResult

Get autocomplete suggestions for a prefix.

Parameters
ParameterTypeDescription
`prefix`strThe prefix to get suggestions for.
`field`strThe field to search for suggestions.
`filters`dict[str, Any] | NoneOptional filters to apply.
Returns
TypeDescription
SuggestionResultSuggestionResult with matching suggestions.
did_you_mean
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
ParameterTypeDescription
`query`strThe misspelled query.
`max_variants`intMaximum number of correction variants to generate.
Returns
TypeDescription
SuggestionResultSuggestionResult with correction suggestions.

Result from a suggestion query.

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]

Raised when search backend encounters an error.

Raised when search configuration is invalid.

Raised when search query execution fails.

Base exception for search operations.

Raised when search index operation fails.