tessl/pypi-algoliasearch

A fully-featured and blazing-fast Python API client to interact with Algolia's search-as-a-service platform.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Recommendations

Name: tessl/pypi-algoliasearch
Author: tessl

AI-powered recommendation engine for suggesting relevant content, products, or related items based on user behavior and content similarity. The Recommend API provides personalized suggestions to enhance user experience and drive engagement.

Capabilities

Client Initialization

Create recommendation clients for both async and sync operations.

class RecommendClient:
    def __init__(
        self,
        app_id: Optional[str] = None,
        api_key: Optional[str] = None,
        transporter: Optional[Transporter] = None,
        config: Optional[RecommendConfig] = None
    ) -> None: ...

    @classmethod
    def create_with_config(
        cls, 
        config: RecommendConfig, 
        transporter: Optional[Transporter] = None
    ) -> RecommendClient: ...

    async def close(self) -> None: ...
    async def set_client_api_key(self, api_key: str) -> None: ...

class RecommendClientSync:
    def __init__(
        self,
        app_id: Optional[str] = None,
        api_key: Optional[str] = None,
        transporter: Optional[TransporterSync] = None,
        config: Optional[RecommendConfig] = None
    ) -> None: ...

Get Recommendations

Retrieve personalized recommendations based on user behavior, content similarity, or custom strategies.

async def get_recommendations(
    self,
    get_recommendations_params: Union[GetRecommendationsParams, dict],
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetRecommendationsResponse:
    """
    Get personalized recommendations for users or items.
    
    Parameters:
    - get_recommendations_params: Recommendation request configuration
    - request_options: Additional request options
    
    Returns:
    GetRecommendationsResponse with recommended items
    """

Recommendation Rules Management

Manage recommendation rules that control how suggestions are generated and filtered.

async def search_recommend_rules(
    self,
    index_name: str,
    search_recommend_rules_params: Optional[Union[SearchRecommendRulesParams, dict]] = None,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchRecommendRulesResponse:
    """
    Search for recommendation rules in an index.
    
    Parameters:
    - index_name: Target index name
    - search_recommend_rules_params: Search parameters for rules
    - request_options: Additional request options
    
    Returns:
    SearchRecommendRulesResponse with matching rules
    """

async def get_recommend_rule(
    self,
    index_name: str,
    model: str,
    object_id: str,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> RecommendRule:
    """
    Get a specific recommendation rule.
    
    Parameters:
    - index_name: Target index name
    - model: Recommendation model identifier
    - object_id: Rule identifier
    - request_options: Additional request options
    
    Returns:
    RecommendRule object
    """

async def create_recommend_rule(
    self,
    index_name: str,
    model: str,
    recommend_rule: Union[RecommendRule, dict],
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> RecommendUpdatedAtResponse:
    """
    Create a new recommendation rule.
    
    Parameters:
    - index_name: Target index name
    - model: Recommendation model identifier
    - recommend_rule: Rule configuration
    - request_options: Additional request options
    
    Returns:
    RecommendUpdatedAtResponse with creation confirmation
    """

async def update_recommend_rule(
    self,
    index_name: str,
    model: str,
    object_id: str,
    recommend_rule: Union[RecommendRule, dict],
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> RecommendUpdatedAtResponse:
    """
    Update an existing recommendation rule.
    
    Parameters:
    - index_name: Target index name
    - model: Recommendation model identifier
    - object_id: Rule identifier
    - recommend_rule: Updated rule configuration
    - request_options: Additional request options
    
    Returns:
    RecommendUpdatedAtResponse with update confirmation
    """

async def delete_recommend_rule(
    self,
    index_name: str,
    model: str,
    object_id: str,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> RecommendUpdatedAtResponse:
    """
    Delete a recommendation rule.
    
    Parameters:
    - index_name: Target index name
    - model: Recommendation model identifier
    - object_id: Rule identifier to delete
    - request_options: Additional request options
    
    Returns:
    RecommendUpdatedAtResponse with deletion confirmation
    """

async def batch_recommend_rules(
    self,
    index_name: str,
    model: str,
    batch_write_params: Union[BatchWriteParams, dict],
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> RecommendUpdatedAtResponse:
    """
    Perform batch operations on recommendation rules.
    
    Parameters:
    - index_name: Target index name
    - model: Recommendation model identifier
    - batch_write_params: Batch operations to perform
    - request_options: Additional request options
    
    Returns:
    RecommendUpdatedAtResponse with batch operation confirmation
    """

async def get_recommend_status(
    self,
    index_name: str,
    model: str,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetRecommendStatusResponse:
    """
    Get the status of recommendation model for an index.
    
    Parameters:
    - index_name: Target index name
    - model: Recommendation model identifier
    - request_options: Additional request options
    
    Returns:
    GetRecommendStatusResponse with model status information
    """

Custom API Endpoints

Execute custom HTTP operations for advanced recommendation features.

async def custom_delete(
    self,
    path: str,
    parameters: Optional[Dict[str, Any]] = None,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> dict:
    """
    Perform a custom DELETE request.
    
    Parameters:
    - path: API endpoint path
    - parameters: Query parameters
    - request_options: Additional request options
    
    Returns:
    Response data as dictionary
    """

async def custom_get(
    self,
    path: str,
    parameters: Optional[Dict[str, Any]] = None,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> dict:
    """
    Perform a custom GET request.
    
    Parameters:
    - path: API endpoint path
    - parameters: Query parameters
    - request_options: Additional request options
    
    Returns:
    Response data as dictionary
    """

async def custom_post(
    self,
    path: str,
    parameters: Optional[Dict[str, Any]] = None,
    body: Optional[Dict[str, Any]] = None,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> dict:
    """
    Perform a custom POST request.
    
    Parameters:
    - path: API endpoint path
    - parameters: Query parameters
    - body: Request body data
    - request_options: Additional request options
    
    Returns:
    Response data as dictionary
    """

async def custom_put(
    self,
    path: str,
    parameters: Optional[Dict[str, Any]] = None,
    body: Optional[Dict[str, Any]] = None,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> dict:
    """
    Perform a custom PUT request.
    
    Parameters:
    - path: API endpoint path
    - parameters: Query parameters
    - body: Request body data
    - request_options: Additional request options
    
    Returns:
    Response data as dictionary
    """

Usage Examples

Basic Recommendations

from algoliasearch.recommend.client import RecommendClient

# Initialize client
client = RecommendClient("YOUR_APP_ID", "YOUR_API_KEY")

# Get related products recommendations
response = await client.get_recommendations({
    "requests": [{
        "indexName": "products",
        "model": "related-products", 
        "objectID": "product-123",
        "maxRecommendations": 10
    }]
})

# Process recommendations
for hit in response.results[0].hits:
    print(f"Recommended: {hit['name']} - Score: {hit.get('_score', 'N/A')}")

Trending Items

# Get trending items
response = await client.get_recommendations({
    "requests": [{
        "indexName": "products",
        "model": "trending-items",
        "maxRecommendations": 20,
        "facetFilters": [
            "category:electronics"
        ]
    }]
})

Frequently Bought Together

# Get frequently bought together recommendations
response = await client.get_recommendations({
    "requests": [{
        "indexName": "products", 
        "model": "bought-together",
        "objectID": "product-456",
        "maxRecommendations": 5
    }]
})

Types

# Recommendation request types
class GetRecommendationsParams(BaseModel):
    requests: List[RecommendationsRequest]

class RecommendationsRequest(BaseModel):
    index_name: str
    model: str
    object_id: Optional[str] = None
    max_recommendations: Optional[int] = None
    query_parameters: Optional[dict] = None
    facet_filters: Optional[List[str]] = None
    numeric_filters: Optional[List[str]] = None
    filters: Optional[str] = None
    user_token: Optional[str] = None

# Response types
class GetRecommendationsResponse(BaseModel):
    results: List[RecommendationsResult]

class RecommendationsResult(BaseModel):
    hits: List[dict]
    nb_hits: int
    processing_time_ms: int
    query: Optional[str] = None
    params: Optional[str] = None

# Rule types  
class RecommendRule(BaseModel):
    object_id: str
    condition: dict
    consequence: dict
    description: Optional[str] = None
    enabled: Optional[bool] = None

class SearchRecommendRulesParams(BaseModel):
    query: Optional[str] = None
    anchoring: Optional[str] = None
    context: Optional[str] = None
    page: Optional[int] = None
    hits_per_page: Optional[int] = None
    enabled: Optional[bool] = None

class SearchRecommendRulesResponse(BaseModel):
    hits: List[RecommendRule]
    nb_hits: int
    page: int
    nb_pages: int
    hits_per_page: int

class RecommendUpdatedAtResponse(BaseModel):
    task_id: int
    updated_at: str

class GetRecommendStatusResponse(BaseModel):
    is_running: bool
    processing_time_ms: Optional[int] = None
    last_built_at: Optional[str] = None
    nb_records: Optional[int] = None