CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-redis

Python client for Redis database and key-value store

—

Pending
Overview
Eval results
Files

redis-search.mddocs/modules/

RediSearch

RediSearch provides full-text search, secondary indexing, and query capabilities for Redis. It supports complex queries, aggregations, auto-complete, and geospatial search with high performance indexing.

Capabilities

Index Management

Create and manage search indexes for Redis data structures.

def ft_create(
    self,
    index_name: str,
    schema: Dict[str, Any],
    definition: Optional[Dict[str, Any]] = None
) -> str: ...

def ft_dropindex(
    self,
    index_name: str,
    delete_documents: bool = False
) -> str: ...

def ft_info(self, index_name: str) -> Dict[str, Any]: ...

def ft_list(self) -> List[str]: ...

def ft_alter(
    self,
    index_name: str,
    schema_add: Dict[str, Any]
) -> str: ...

def ft_aliasadd(self, alias: str, index_name: str) -> str: ...

def ft_aliasdel(self, alias: str) -> str: ...

def ft_aliasupdate(self, alias: str, index_name: str) -> str: ...

Search Operations

Perform full-text search and complex queries on indexed data.

def ft_search(
    self,
    index_name: str,
    query: str,
    query_params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]: ...

def ft_aggregate(
    self,
    index_name: str,
    query: str,
    *args: Any
) -> List[Any]: ...

def ft_explain(
    self,
    index_name: str,
    query: str
) -> str: ...

def ft_explaincli(
    self,
    index_name: str,
    query: str
) -> str: ...

def ft_profile(
    self,
    index_name: str,
    query: str,
    limited: bool = False
) -> List[Any]: ...

Document Operations

Add, update, and delete documents in search indexes.

def ft_add(
    self,
    index_name: str,
    doc_id: str,
    score: float,
    fields: Dict[str, Any],
    **kwargs
) -> str: ...

def ft_del(
    self,
    index_name: str,
    doc_id: str,
    delete_document: bool = False
) -> int: ...

def ft_get(
    self,
    index_name: str,
    *doc_ids: str
) -> List[Optional[Dict[str, Any]]]: ...

def ft_mget(
    self,
    index_name: str,
    *doc_ids: str
) -> List[Optional[Dict[str, Any]]]: ...

Suggestion and Auto-complete

Manage suggestion dictionaries for auto-complete functionality.

def ft_sugadd(
    self,
    key: str,
    string: str,
    score: float,
    incr: bool = False,
    payload: Optional[str] = None
) -> int: ...

def ft_sugget(
    self,
    key: str,
    prefix: str,
    fuzzy: bool = False,
    max_results: int = 5,
    with_scores: bool = False,
    with_payloads: bool = False
) -> List[Any]: ...

def ft_sugdel(self, key: str, string: str) -> int: ...

def ft_suglen(self, key: str) -> int: ...

Dictionary Management

Manage custom dictionaries for spell checking and synonyms.

def ft_dictadd(self, dictionary: str, *terms: str) -> int: ...

def ft_dictdel(self, dictionary: str, *terms: str) -> int: ...

def ft_dictdump(self, dictionary: str) -> List[str]: ...

Configuration

Configure RediSearch module settings.

def ft_config_get(self, option: str) -> Dict[str, Any]: ...

def ft_config_set(self, option: str, value: Any) -> str: ...

Usage Examples

Creating Search Indexes

import redis
from redis.commands.search import Search
from redis.commands.search.field import TextField, NumericField, TagField, GeoField

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Create index for product catalog
def create_product_index():
    try:
        # Define schema for products
        schema = [
            TextField("title", weight=5.0),
            TextField("description", weight=1.0),
            TagField("category"),
            TagField("brand"),
            NumericField("price"),
            NumericField("rating"),
            TagField("tags", separator=","),
            GeoField("location")
        ]
        
        # Create index
        result = r.ft("products").create_index(
            schema,
            definition={
                'prefix': ['product:'],
                'language': 'english'
            }
        )
        print(f"Created index: {result}")
        
    except Exception as e:
        print(f"Index creation error: {e}")

create_product_index()

# Add sample products
products = [
    {
        "id": "product:1001",
        "title": "Gaming Laptop",
        "description": "High-performance laptop for gaming and professional work",
        "category": "Electronics",
        "brand": "TechBrand",
        "price": 1299.99,
        "rating": 4.5,
        "tags": "gaming,laptop,performance",
        "location": "40.7128,-74.0060"  # NYC coordinates
    },
    {
        "id": "product:1002", 
        "title": "Wireless Headphones",
        "description": "Premium noise-canceling wireless headphones",
        "category": "Electronics",
        "brand": "AudioPro",
        "price": 299.99,
        "rating": 4.2,
        "tags": "audio,wireless,headphones",
        "location": "34.0522,-118.2437"  # LA coordinates
    }
]

for product in products:
    product_id = product.pop("id")
    r.hset(product_id, mapping=product)

print("Added sample products to Redis")

Basic Search Operations

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Simple text search
def search_products(query):
    try:
        result = r.ft("products").search(query)
        print(f"Search '{query}' found {result.total} results:")
        
        for doc in result.docs:
            print(f"  ID: {doc.id}")
            print(f"  Title: {doc.title}")
            print(f"  Price: ${doc.price}")
            print(f"  Rating: {doc.rating}")
            print()
            
    except Exception as e:
        print(f"Search error: {e}")

# Search for gaming products
search_products("gaming")

# Search in specific field
search_products("@title:laptop")

# Search with multiple terms
search_products("wireless headphones")

# Phrase search
search_products('"gaming laptop"')

Advanced Search Queries

import redis
from redis.commands.search.query import Query

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Complex search with filters
def advanced_product_search():
    # Price range filter
    query = Query("@category:Electronics @price:[200 500]").return_fields("title", "price", "rating")
    result = r.ft("products").search(query)
    print(f"Electronics $200-$500: {result.total} results")
    
    # Rating filter with sorting
    query = Query("@rating:[4.0 5.0]").sort_by("price", asc=True).return_fields("title", "price", "rating")
    result = r.ft("products").search(query)
    print(f"High-rated products (sorted by price): {result.total} results")
    
    # Tag search
    query = Query("@tags:{gaming}").return_fields("title", "tags")
    result = r.ft("products").search(query)
    print(f"Gaming tagged products: {result.total} results")
    
    # Geospatial search (within 1000km of NYC)
    query = Query("@location:[40.7128 -74.0060 1000 km]").return_fields("title", "location")
    result = r.ft("products").search(query)
    print(f"Products near NYC: {result.total} results")
    
    # Boolean search
    query = Query("(@title:laptop) | (@title:headphones)").return_fields("title", "category")
    result = r.ft("products").search(query)
    print(f"Laptops OR headphones: {result.total} results")

advanced_product_search()

Search Aggregations

import redis
from redis.commands.search.aggregation import AggregateRequest
from redis.commands.search import reducers

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def product_analytics():
    # Average price by category
    agg_request = AggregateRequest("*").group_by(
        "@category",
        reducers.avg("@price").alias("avg_price"),
        reducers.count().alias("count")
    )
    
    result = r.ft("products").aggregate(agg_request)
    print("Average price by category:")
    for row in result.rows:
        print(f"  {row[1]}: ${float(row[3]):.2f} (count: {row[5]})")
    
    # Top brands by average rating
    agg_request = AggregateRequest("*").group_by(
        "@brand",
        reducers.avg("@rating").alias("avg_rating"),
        reducers.count().alias("product_count")
    ).sort_by("@avg_rating", desc=True)
    
    result = r.ft("products").aggregate(agg_request)
    print("\nTop brands by rating:")
    for row in result.rows:
        print(f"  {row[1]}: {float(row[3]):.2f} ({row[5]} products)")

product_analytics()

Auto-complete and Suggestions

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def setup_autocomplete():
    # Add product titles to suggestion dictionary
    suggestions = [
        ("Gaming Laptop", 100, "product:1001"),
        ("Wireless Headphones", 90, "product:1002"),
        ("Gaming Mouse", 80, "product:1003"),
        ("Laptop Stand", 70, "product:1004"),
        ("Wireless Keyboard", 85, "product:1005")
    ]
    
    for suggestion, score, payload in suggestions:
        r.ft().sugadd("product_suggestions", suggestion, score, payload=payload)
    
    print("Added product suggestions")

def test_autocomplete(prefix):
    # Get suggestions for prefix
    suggestions = r.ft().sugget(
        "product_suggestions",
        prefix,
        fuzzy=True,
        max=5,
        with_scores=True,
        with_payloads=True
    )
    
    print(f"Suggestions for '{prefix}':")
    for suggestion in suggestions:
        if len(suggestion) >= 3:
            text, score, payload = suggestion[0], suggestion[1], suggestion[2]
            print(f"  {text} (score: {score}, product: {payload})")
        else:
            print(f"  {suggestion[0]}")

setup_autocomplete()
test_autocomplete("gam")
test_autocomplete("lap")
test_autocomplete("wire")

Full-Text Search with Highlighting

import redis
from redis.commands.search.query import Query

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def search_with_highlighting(search_term):
    # Search with result highlighting
    query = Query(search_term).highlight(
        fields=["title", "description"],
        tags=["<b>", "</b>"]
    ).return_fields("title", "description", "price")
    
    result = r.ft("products").search(query)
    
    print(f"Search results for '{search_term}' with highlighting:")
    for doc in result.docs:
        print(f"Title: {doc.title}")
        print(f"Description: {doc.description}")
        print(f"Price: ${doc.price}")
        print("-" * 50)

search_with_highlighting("gaming laptop")
search_with_highlighting("wireless")

Spell Checking and Synonyms

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def setup_dictionaries():
    # Add terms to custom dictionary for spell checking
    tech_terms = ["laptop", "desktop", "smartphone", "tablet", "headphones"]
    r.ft().dictadd("tech_dict", *tech_terms)
    
    # Add synonyms
    synonyms = ["laptop", "notebook", "computer"]
    r.ft().dictadd("laptop_synonyms", *synonyms)
    
    print("Added custom dictionaries")

def spell_check_search(query):
    # This would typically be implemented with a custom spell checker
    # using the dictionaries created above
    try:
        result = r.ft("products").search(query)
        if result.total == 0:
            print(f"No results for '{query}'. Did you mean:")
            # Implement fuzzy search or suggestion logic here
            fuzzy_query = Query(query).no_content()
            fuzzy_result = r.ft("products").search(fuzzy_query)
            print(f"Found {fuzzy_result.total} potential matches")
    except Exception as e:
        print(f"Search error: {e}")

setup_dictionaries()
spell_check_search("laptpo")  # Misspelled "laptop"

Index Management and Monitoring

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def monitor_search_indexes():
    # List all indexes
    indexes = r.ft()._list()
    print(f"Available indexes: {indexes}")
    
    # Get detailed index information
    for index_name in indexes:
        try:
            info = r.ft(index_name).info()
            print(f"\nIndex: {index_name}")
            print(f"  Documents: {info.get('num_docs', 'N/A')}")
            print(f"  Terms: {info.get('num_terms', 'N/A')}")
            print(f"  Records: {info.get('num_records', 'N/A')}")
            print(f"  Index size: {info.get('inverted_sz_mb', 'N/A')} MB")
            
        except Exception as e:
            print(f"Error getting info for {index_name}: {e}")

def explain_query(query):
    # Explain query execution plan
    try:
        explanation = r.ft("products").explain(query)
        print(f"Query execution plan for '{query}':")
        print(explanation)
    except Exception as e:
        print(f"Error explaining query: {e}")

monitor_search_indexes()
explain_query("@title:gaming @price:[1000 2000]")

Search Performance Profiling

import redis
from redis.commands.search.query import Query

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def profile_search_performance():
    queries = [
        "gaming",
        "@title:laptop",
        "@price:[200 500]",
        "(@title:gaming) | (@category:Electronics)"
    ]
    
    for query in queries:
        try:
            # Profile query performance
            profile_result = r.ft("products").profile(Query(query))
            
            print(f"\nQuery: {query}")
            print(f"Profile results: {profile_result}")
            
        except Exception as e:
            print(f"Error profiling query '{query}': {e}")

profile_search_performance()

Batch Document Operations

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def batch_document_operations():
    # Batch add documents to index
    documents = [
        {
            "id": "product:2001",
            "title": "Gaming Mouse",
            "description": "RGB gaming mouse with programmable buttons",
            "category": "Electronics",
            "brand": "GamePro",
            "price": 79.99,
            "rating": 4.3
        },
        {
            "id": "product:2002",
            "title": "Mechanical Keyboard",
            "description": "Cherry MX switches mechanical keyboard",
            "category": "Electronics", 
            "brand": "KeyMaster",
            "price": 149.99,
            "rating": 4.7
        }
    ]
    
    # Use pipeline for batch operations
    pipe = r.pipeline()
    
    for doc in documents:
        doc_id = doc.pop("id")
        pipe.hset(doc_id, mapping=doc)
    
    results = pipe.execute()
    print(f"Batch added {len(results)} documents")
    
    # Batch retrieve documents
    doc_ids = ["product:2001", "product:2002"]
    retrieved_docs = r.ft("products").mget(*doc_ids)
    
    print("Retrieved documents:")
    for doc in retrieved_docs:
        if doc:
            print(f"  {doc.get('title', 'N/A')} - ${doc.get('price', 'N/A')}")

batch_document_operations()

Install with Tessl CLI

npx tessl i tessl/pypi-redis

docs

modules

redis-bloom.md

redis-json.md

redis-search.md

redis-timeseries.md

async-support.md

cluster-support.md

connection-management.md

core-client.md

distributed-locking.md

error-handling.md

high-availability.md

index.md

pipelines-transactions.md

pubsub-messaging.md

tile.json