A fully-featured and blazing-fast Python API client to interact with Algolia's search-as-a-service platform.
npx @tessl/cli install tessl/pypi-algoliasearch@4.26.0A fully-featured and blazing-fast Python API client to interact with Algolia's search-as-a-service platform. This comprehensive client provides access to Algolia's full suite of services including search, recommendations, analytics, A/B testing, data ingestion, insights, monitoring, personalization, and query suggestions.
pip install 'algoliasearch>=4.0,<5.0'from algoliasearch.search.client import SearchClientFor other services:
from algoliasearch.recommend.client import RecommendClient
from algoliasearch.analytics.client import AnalyticsClient
from algoliasearch.insights.client import InsightsClient
from algoliasearch.abtesting.client import AbtestingClient
from algoliasearch.ingestion.client import IngestionClient
from algoliasearch.monitoring.client import MonitoringClient
from algoliasearch.personalization.client import PersonalizationClient
from algoliasearch.query_suggestions.client import QuerySuggestionsClient
from algoliasearch.composition.client import CompositionClientfrom algoliasearch.search.client import SearchClient
# Initialize client
client = SearchClient("YOUR_APP_ID", "YOUR_API_KEY")
# Save an object to an index
response = await client.save_object(
index_name="products",
body={
"objectID": "123",
"name": "Blue T-Shirt",
"brand": "Fashion Co",
"price": 29.99
}
)
# Wait for the indexing task to complete
await client.wait_for_task(index_name="products", task_id=response.task_id)
# Search for products
response = await client.search(
search_method_params={
"requests": [{
"indexName": "products",
"query": "blue shirt",
"hitsPerPage": 10
}]
}
)
# Print results
for hit in response.results[0].hits:
print(f"{hit['name']} - ${hit['price']}")The Algoliasearch Python client is built with a service-oriented architecture:
Core search functionality including index management, object operations, search queries, rules, synonyms, and API key management. This is the primary interface for most Algolia operations.
class SearchClient:
def __init__(self, app_id: str = None, api_key: str = None): ...
async def search(self, search_method_params: SearchMethodParams) -> SearchResponse: ...
async def save_object(self, index_name: str, body: dict) -> SaveObjectResponse: ...
async def save_objects(self, index_name: str, objects: List[dict]) -> BatchResponse: ...
async def delete_object(self, index_name: str, object_id: str) -> DeletedAtResponse: ...
async def get_object(self, index_name: str, object_id: str) -> dict: ...
async def browse_objects(self, index_name: str, browse_params: BrowseParams = None) -> BrowseResponse: ...AI-powered recommendation engine for suggesting relevant content, products, or related items based on user behavior and content similarity.
class RecommendClient:
def __init__(self, app_id: str = None, api_key: str = None): ...
async def get_recommendations(self, get_recommendations_params: GetRecommendationsParams) -> GetRecommendationsResponse: ...
async def search_recommend_rules(self, index_name: str, search_recommend_rules_params: SearchRecommendRulesParams = None) -> SearchRecommendRulesResponse: ...Comprehensive analytics for search performance, user behavior tracking, and business insights including click-through rates, conversion metrics, and user interaction data.
class AnalyticsClient:
def __init__(self, app_id: str = None, api_key: str = None): ...
async def get_click_through_rate(self, index: str, **kwargs) -> GetClickThroughRateResponse: ...
async def get_conversion_rate(self, index: str, **kwargs) -> GetConversionRateResponse: ...
async def get_search_volume(self, index: str, **kwargs) -> GetSearchVolumeResponse: ...
class InsightsClient:
def __init__(self, app_id: str = None, api_key: str = None): ...
async def push_events(self, insights_events: InsightsEvents) -> EventsResponse: ...
async def click_through_events(self, click_through_events: ClickThroughEvents) -> EventsResponse: ...Controlled experimentation platform for testing search configurations, UI changes, and business strategies with statistical significance.
class AbtestingClient:
def __init__(self, app_id: str = None, api_key: str = None): ...
async def list_ab_tests(self, **kwargs) -> ListABTestsResponse: ...
async def get_ab_test(self, id: int) -> ABTest: ...
async def add_ab_test(self, ab_test_create: ABTestCreate) -> ABTestResponse: ...
async def stop_ab_test(self, id: int) -> ABTestResponse: ...Scalable data pipeline services for importing, transforming, and synchronizing large datasets from various sources into Algolia indices.
class IngestionClient:
def __init__(self, app_id: str = None, api_key: str = None): ...
async def list_sources(self, **kwargs) -> ListSourcesResponse: ...
async def create_source(self, source_create: SourceCreate) -> Source: ...
async def list_transformations(self, **kwargs) -> ListTransformationsResponse: ...
async def create_transformation(self, transformation_create: TransformationCreate) -> Transformation: ...Service health monitoring, infrastructure metrics, and operational management tools for maintaining optimal search performance.
class MonitoringClient:
def __init__(self, app_id: str = None, api_key: str = None): ...
async def get_cluster_status(self, **kwargs) -> ClusterStatus: ...
async def get_infrastructure_metrics(self, **kwargs) -> InfrastructureMetrics: ...
class PersonalizationClient:
def __init__(self, app_id: str = None, api_key: str = None): ...
async def get_personalization_strategy(self, **kwargs) -> PersonalizationStrategy: ...
async def set_personalization_strategy(self, personalization_strategy_params: PersonalizationStrategyParams) -> SetPersonalizationStrategyResponse: ...Intelligent query completion and suggestion services that help users discover relevant search terms and improve search experience.
class QuerySuggestionsClient:
def __init__(self, app_id: str = None, api_key: str = None): ...
async def list_configs(self, **kwargs) -> ListConfigsResponse: ...
async def create_config(self, configuration_with_index: ConfigurationWithIndex) -> BaseResponse: ...
async def get_config_status(self, index_name: str) -> GetConfigStatusResponse: ...from typing import Dict, List, Optional, Union, Any
from pydantic import BaseModel
# Core response types
class ApiResponse(BaseModel):
def deserialize(self, response_type: type, data: str): ...
def to_json(self) -> str: ...
# Request options for all methods
class RequestOptions(BaseModel):
headers: Optional[Dict[str, str]] = None
query_params: Optional[Dict[str, Any]] = None
timeout: Optional[int] = None
# Authentication and configuration
class BaseConfig(BaseModel):
app_id: str
api_key: str
hosts: Optional[List[str]] = None
# Task tracking for async operations
class GetTaskResponse(BaseModel):
task_id: int
status: str
pending_task: bool
# Common batch operation response
class BatchResponse(BaseModel):
task_id: int
object_ids: List[str]