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

Monitoring and Management

Name: tessl/pypi-algoliasearch
Author: tessl

Service health monitoring, infrastructure metrics, operational management tools, and personalization services for maintaining optimal search performance and user experience.

Capabilities

Monitoring Client

Monitor service health and infrastructure performance.

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

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

Infrastructure Monitoring

Monitor cluster status, server health, and performance metrics.

async def get_cluster_status(
    self,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> ClusterStatus:
    """
    Get current cluster health and status.
    
    Returns:
    ClusterStatus with health information
    """

async def get_infrastructure_metrics(
    self,
    metric: str,
    start_date: str,
    end_date: str,
    request_options: Optional[Union[dict, RequestOptions]] = None  
) -> InfrastructureMetrics:
    """
    Get infrastructure performance metrics.
    
    Parameters:
    - metric: Metric type to retrieve
    - start_date: Start date (YYYY-MM-DD format)
    - end_date: End date (YYYY-MM-DD format)
    - request_options: Additional request options
    
    Returns:
    InfrastructureMetrics with performance data
    """

async def get_servers(
    self,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> ServersResponse:
    """
    Get list of available servers and their status.
    
    Returns:
    ServersResponse with server information
    """

Personalization Client

Manage search personalization and user experience optimization.

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

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

Personalization Strategy

Configure and manage personalization strategies for enhanced user experience.

async def get_personalization_strategy(
    self,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> PersonalizationStrategy:
    """
    Get current personalization strategy configuration.
    
    Returns:
    PersonalizationStrategy with current settings
    """

async def set_personalization_strategy(
    self,
    personalization_strategy_params: Union[PersonalizationStrategyParams, dict],
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> SetPersonalizationStrategyResponse:
    """
    Configure personalization strategy settings.
    
    Parameters:
    - personalization_strategy_params: Strategy configuration
    - request_options: Additional request options
    
    Returns:
    SetPersonalizationStrategyResponse with update confirmation
    """

async def delete_user_profile(
    self,
    user_token: str,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> dict:
    """
    Delete a user's personalization profile.
    
    Parameters:
    - user_token: Unique user identifier
    - request_options: Additional request options
    
    Returns:
    Deletion confirmation response
    """

async def get_user_token_profile(
    self,
    user_token: str,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetUserTokenProfileResponse:
    """
    Get personalization profile for a specific user.
    
    Parameters:
    - user_token: Unique user identifier
    - request_options: Additional request options
    
    Returns:
    GetUserTokenProfileResponse with user profile data
    """

Composition Client

Manage search compositions for advanced query orchestration.

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

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

Composition Operations

Execute complex search operations across multiple indices and sources.

async def search(
    self,
    composition_id: str,
    request_body: Union[RequestBody, dict],
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchResponse:
    """
    Execute search on a composition.
    
    Parameters:
    - composition_id: Unique composition identifier
    - request_body: Search request body
    - request_options: Additional request options
    
    Returns:
    SearchResponse with composition search results
    """

async def search_for_facet_values(
    self,
    composition_id: str,
    facet_name: str,
    search_for_facet_values_request: Optional[Union[SearchForFacetValuesRequest, dict]] = None,
    request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchForFacetValuesResponse:
    """
    Search for facet values within a composition.
    
    Parameters:
    - composition_id: Unique composition identifier
    - facet_name: Facet attribute name
    - search_for_facet_values_request: Facet search parameters
    - request_options: Additional request options
    
    Returns:
    SearchForFacetValuesResponse with facet values
    """

Usage Examples

Monitor Infrastructure

from algoliasearch.monitoring.client import MonitoringClient

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

# Check cluster status
status = await client.get_cluster_status()
print(f"Cluster status: {status.status}")
print(f"Last update: {status.last_update_timestamp}")

# Get infrastructure metrics
metrics = await client.get_infrastructure_metrics(
    metric="cpu_usage",
    start_date="2024-01-01",
    end_date="2024-01-31"
)

Configure Personalization

from algoliasearch.personalization.client import PersonalizationClient

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

# Set personalization strategy
strategy_response = await client.set_personalization_strategy({
    "eventsScoring": [
        {
            "eventName": "Add to cart",
            "eventType": "conversion", 
            "score": 50
        },
        {
            "eventName": "Purchase",
            "eventType": "conversion",
            "score": 100
        }
    ],
    "facetsScoring": [
        {
            "facetName": "brand",
            "score": 100
        },
        {
            "facetName": "category", 
            "score": 10
        }
    ],
    "personalizationImpact": 95
})

print(f"Strategy updated at: {strategy_response.updated_at}")

Types

# Monitoring types
class ClusterStatus(BaseModel):
    status: str
    last_update_timestamp: str
    degraded: bool

class InfrastructureMetrics(BaseModel):
    metric_name: str
    data_points: List[dict]
    start_date: str
    end_date: str

class ServersResponse(BaseModel):
    servers: List[dict]

# Personalization types  
class PersonalizationStrategy(BaseModel):
    events_scoring: List[EventScoring]
    facets_scoring: List[FacetScoring]
    personalization_impact: int

class EventScoring(BaseModel):
    event_name: str
    event_type: str
    score: int

class FacetScoring(BaseModel):
    facet_name: str
    score: int

class PersonalizationStrategyParams(BaseModel):
    events_scoring: Optional[List[EventScoring]] = None
    facets_scoring: Optional[List[FacetScoring]] = None
    personalization_impact: Optional[int] = None

class SetPersonalizationStrategyResponse(BaseModel):
    status: int
    updated_at: str

class GetUserTokenProfileResponse(BaseModel):
    user_token: str
    last_event_at: str
    scores: Optional[dict] = None

# Composition types
class RequestBody(BaseModel):
    query: Optional[str] = None
    params: Optional[dict] = None

class SearchResponse(BaseModel):
    hits: List[dict]
    nb_hits: int
    processing_time_ms: int
    query: str

Install with Tessl CLI