beanis.odm.indexes

Redis-based indexing system using Sets and Sorted Sets

This module provides secondary indexing capabilities for Beanis documents using native Redis data structures: - Sets for categorical/string fields (exact match lookups) - Sorted Sets for numeric fields (range queries)

GeoPoint

class GeoPoint(BaseModel)

Represents a geographic point with longitude and latitude

Usage: class Store(Document): name: str location: Indexed[GeoPoint] # Geo index

store = Store(name="HQ", location=GeoPoint(longitude=-122.4, latitude=37.8))

GeoPoint.as_tuple

def as_tuple() -> Tuple[float, float]

Return (longitude, latitude) tuple for Redis GEOADD

IndexType

class IndexType()

Types of indexes supported

IndexType.SET

For categorical/string fields (exact match)

IndexType.SORTED_SET

For numeric fields (range queries)

IndexType.GEO

For geo-spatial fields (location-based queries)

IndexType.VECTOR

For vector similarity search (embeddings)

IndexedField

class IndexedField()

Marks a field as indexed for secondary index support

Usage: class Product(Document): category: Annotated[str, IndexedField()] # Set index price: Annotated[float, IndexedField()] # Sorted Set index location: Annotated[GeoPoint, IndexedField()] # Geo index

IndexedField.__init__

def __init__(index_type: Optional[str] = None)

Arguments:

  • index_type: Type of index ("set", "zset", or "geo"). If None, auto-detect based on field type

VectorField

def VectorField(dimensions: int,
                algorithm: str = "HNSW",
                distance_metric: str = "COSINE",
                m: int = 16,
                ef_construction: int = 200) -> IndexedField

Helper to create a vector field for similarity search

Usage: class Document(Document): embedding: Annotated[List[float], VectorField(dimensions=1024)]

Arguments:

  • dimensions - Vector dimensionality (e.g., 1024 for Jina v4)
  • algorithm - "HNSW" (default) or "FLAT"
  • distance_metric - "COSINE" (default), "L2", or "IP" (inner product)
  • m - HNSW M parameter (connections per node)
  • ef_construction - HNSW ef_construction parameter

IndexManager

class IndexManager()

Manages secondary indexes for documents using Redis Sets and Sorted Sets

IndexManager.get_index_key

@staticmethod
def get_index_key(document_class: Type,
                  field_name: str,
                  value: Any = None) -> str

Generate Redis key for an index

For Set indexes: idx:Product:category:electronics For Sorted Set indexes: idx:Product:price

IndexManager.get_indexed_fields

@staticmethod
def get_indexed_fields(document_class: Type) -> Dict[str, IndexedField]

Extract all indexed fields from a document class

Returns dict: {field_name: IndexedField}

IndexManager.determine_index_type

@staticmethod
def determine_index_type(document_class: Type, field_name: str,
                         indexed_field: IndexedField) -> str

Determine the index type based on field type

  • Numeric types (int, float) -> Sorted Set (zset)
  • GeoPoint types -> Geo index
  • String/categorical types -> Set

IndexManager.add_to_index

@staticmethod
async def add_to_index(redis_client, document_class: Type, document_id: str,
                       field_name: str, value: Any, index_type: str)

Add document ID to the appropriate index

IndexManager.remove_from_index

@staticmethod
async def remove_from_index(redis_client, document_class: Type,
                            document_id: str, field_name: str, value: Any,
                            index_type: str)

Remove document ID from the appropriate index

IndexManager.update_indexes

@staticmethod
async def update_indexes(redis_client, document_class: Type, document_id: str,
                         old_values: Optional[Dict[str, Any]],
                         new_values: Dict[str, Any])

Update all indexes when a document changes

Uses Redis pipeline for batch operations (performance optimization)

Arguments:

  • old_values: Previous field values (for removal from old indexes)
  • new_values: New field values (for adding to new indexes)

IndexManager.remove_all_indexes

@staticmethod
async def remove_all_indexes(redis_client, document_class: Type,
                             document_id: str, values: Dict[str, Any])

Remove document from all indexes (for deletion) Uses Redis pipeline for batch operations (performance optimization)

IndexManager.find_by_index

@staticmethod
async def find_by_index(redis_client,
                        document_class: Type,
                        field_name: str,
                        value: Any = None,
                        min_value: Any = None,
                        max_value: Any = None) -> List[str]

Find document IDs using an index

For Set indexes (categorical): find_by_index(redis, Product, "category", value="electronics")

For Sorted Set indexes (numeric range): find_by_index(redis, Product, "price", min_value=10, max_value=100)

IndexManager.find_by_geo_radius

@staticmethod
async def find_by_geo_radius(redis_client,
                             document_class: Type,
                             field_name: str,
                             longitude: float,
                             latitude: float,
                             radius: float,
                             unit: str = "km") -> List[str]

Find document IDs within a radius of a geo location

Arguments:

  • field_name: Name of the geo-indexed field
  • longitude: Center point longitude
  • latitude: Center point latitude
  • radius: Search radius
  • unit: Distance unit - 'm', 'km', 'mi', 'ft' (default: 'km') Usage: nearby = await IndexManager.find_by_geo_radius( redis_client, Store, "location", longitude=-122.4, latitude=37.8, radius=10, unit="km" )

IndexManager.find_by_geo_radius_with_distance

@staticmethod
async def find_by_geo_radius_with_distance(
        redis_client,
        document_class: Type,
        field_name: str,
        longitude: float,
        latitude: float,
        radius: float,
        unit: str = "km") -> List[Tuple[str, float]]

Find document IDs within a radius with their distances

Returns list of (document_id, distance) tuples

Usage: nearby = await IndexManager.find_by_geo_radius_with_distance( redis_client, Store, "location", longitude=-122.4, latitude=37.8, radius=10, unit="km" ) for doc_id, distance in nearby: print(f"{doc_id}: {distance} km away")

IndexManager.find_by_vector_similarity

@staticmethod
async def find_by_vector_similarity(
        redis_client,
        document_class: Type,
        field_name: str,
        query_vector: List[float],
        k: int = 10,
        ef_runtime: Optional[int] = None) -> List[Tuple[str, float]]

Find document IDs by vector similarity (KNN search)

Returns list of (document_id, similarity_score) tuples sorted by similarity

Usage: query_embedding = model.encode(["search text"])[0].tolist() results = await IndexManager.find_by_vector_similarity( redis_client, Document, "embedding", query_vector=query_embedding, k=5 ) for doc_id, score in results: doc = await Document.get(doc_id) print(f"{doc.text}: {score}")

Arguments:

  • redis_client - Redis client instance
  • document_class - Document class to search
  • field_name - Name of the vector field
  • query_vector - Query vector as list of floats
  • k - Number of results to return
  • ef_runtime - HNSW ef_runtime parameter (optional, for tuning)

Indexed

def Indexed(field_type: Type, **kwargs) -> Type

Helper function to create an indexed field

Usage: class Product(Document): category: Indexed[str] # Set index price: Indexed[float] # Sorted Set index