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
search-operations.mddocs/
Search Operations
Core search functionality providing comprehensive index management, object operations, search queries, rules, synonyms, API key management, and advanced features like user ID mapping and dictionary management. This is the primary interface for most Algolia operations.
Capabilities
Client Initialization
Create search clients for both async and sync operations.
class SearchClient:
def __init__(
self,
app_id: Optional[str] = None,
api_key: Optional[str] = None,
transporter: Optional[Transporter] = None,
config: Optional[SearchConfig] = None
) -> None: ...
@classmethod
def create_with_config(
cls,
config: SearchConfig,
transporter: Optional[Transporter] = None
) -> SearchClient: ...
async def close(self) -> None: ...
async def set_client_api_key(self, api_key: str) -> None: ...
class SearchClientSync:
def __init__(
self,
app_id: Optional[str] = None,
api_key: Optional[str] = None,
transporter: Optional[TransporterSync] = None,
config: Optional[SearchConfig] = None
) -> None: ...Search Operations
Execute search queries across indices with full-text search, filtering, faceting, and ranking options.
async def search(
self,
search_method_params: Union[SearchMethodParams, dict]
) -> SearchResponse:
"""
Execute search queries across one or more indices.
Parameters:
- search_method_params: Search configuration with requests array
Returns:
SearchResponse with results for each request
"""
async def browse(
self,
index_name: str,
browse_params: Optional[Union[BrowseParams, dict]] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> BrowseResponse:
"""
Browse all objects in an index without search query limitations.
Parameters:
- index_name: Target index name
- browse_params: Browse configuration options
- request_options: Additional request options
Returns:
BrowseResponse with object data and cursor for pagination
"""
async def browse_objects(
self,
index_name: str,
aggregator: Callable[[BrowseResponse], None],
browse_params: Optional[BrowseParamsObject] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> BrowseResponse:
"""
Browse all objects in an index with automatic pagination handling.
Parameters:
- index_name: Target index name
- aggregator: Function to process each page of results
- browse_params: Browse configuration options
- request_options: Additional request options
Returns:
BrowseResponse with final page data
"""
async def search_single_index(
self,
index_name: str,
search_params: Optional[Union[SearchParams, dict]] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchResponse:
"""
Search within a single index.
Parameters:
- index_name: Target index name
- search_params: Search configuration
- request_options: Additional request options
Returns:
SearchResponse with search results
"""
async def search_for_facet_values(
self,
index_name: str,
facet_name: str,
search_for_facet_values_request: Optional[Union[SearchForFacetValuesRequest, dict]] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchForFacetValuesResponse:
"""
Search for values within a specific facet attribute.
Parameters:
- index_name: Target index name
- facet_name: Name of facet attribute to search
- search_for_facet_values_request: Search parameters
- request_options: Additional request options
Returns:
SearchForFacetValuesResponse with matching facet values
"""Object Management
Create, read, update, and delete operations for index objects.
async def save_object(
self,
index_name: str,
body: Union[dict, Any],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SaveObjectResponse:
"""
Save a single object to an index.
Parameters:
- index_name: Target index name
- body: Object data with objectID
- request_options: Additional request options
Returns:
SaveObjectResponse with task information
"""
async def save_objects(
self,
index_name: str,
objects: List[Union[dict, Any]],
wait_for_tasks: bool = False,
batch_size: int = 1000,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> List[BatchResponse]:
"""
Save multiple objects to an index with automatic chunking.
Parameters:
- index_name: Target index name
- objects: List of objects to save
- wait_for_tasks: Wait for all tasks to complete
- batch_size: Objects per batch
- request_options: Additional request options
Returns:
List of BatchResponse objects for each chunk
"""
async def get_objects(
self,
get_objects_params: Union[GetObjectsParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetObjectsResponse:
"""
Retrieve multiple objects by their objectIDs.
Parameters:
- get_objects_params: Parameters with requests array
- request_options: Additional request options
Returns:
GetObjectsResponse with retrieved objects
"""
async def get_object(
self,
index_name: str,
object_id: str,
attributes_to_retrieve: Optional[List[str]] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> dict:
"""
Retrieve a single object by its objectID with optional attribute filtering.
Parameters:
- index_name: Target index name
- object_id: Unique object identifier
- attributes_to_retrieve: Specific attributes to retrieve
- request_options: Additional request options
Returns:
Object data as dictionary
"""
async def delete_object(
self,
index_name: str,
object_id: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> DeletedAtResponse:
"""
Delete a single object from an index.
Parameters:
- index_name: Target index name
- object_id: Unique object identifier
- request_options: Additional request options
Returns:
DeletedAtResponse with task information
"""
async def delete_objects(
self,
index_name: str,
object_ids: List[str],
wait_for_tasks: bool = False,
batch_size: int = 1000,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> List[BatchResponse]:
"""
Delete multiple objects from an index with automatic chunking.
Parameters:
- index_name: Target index name
- object_ids: List of object identifiers to delete
- wait_for_tasks: Wait for all tasks to complete
- batch_size: Objects per batch
- request_options: Additional request options
Returns:
List of BatchResponse objects for each chunk
"""
async def delete_by(
self,
index_name: str,
delete_by_params: Union[DeleteByParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Delete objects matching specific filters.
Parameters:
- index_name: Target index name
- delete_by_params: Deletion criteria and filters
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""
async def partial_update_object(
self,
index_name: str,
object_id: str,
attribute_to_update: Union[AttributeToUpdate, dict],
create_if_not_exists: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtWithObjectIdResponse:
"""
Partially update an object's attributes.
Parameters:
- index_name: Target index name
- object_id: Unique object identifier
- attribute_to_update: Attributes to update
- create_if_not_exists: Create object if it doesn't exist
- request_options: Additional request options
Returns:
UpdatedAtWithObjectIdResponse with task information
"""
async def partial_update_objects(
self,
index_name: str,
objects: List[Union[dict, Any]],
create_if_not_exists: Optional[bool] = None,
wait_for_tasks: bool = False,
batch_size: int = 1000,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> List[BatchResponse]:
"""
Partially update multiple objects' attributes.
Parameters:
- index_name: Target index name
- objects: List of objects with updates
- create_if_not_exists: Create objects if they don't exist
- wait_for_tasks: Wait for all tasks to complete
- batch_size: Objects per batch
- request_options: Additional request options
Returns:
List of BatchResponse objects for each chunk
"""
async def add_or_update_object(
self,
index_name: str,
object_id: str,
body: Dict[str, Any],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtWithObjectIdResponse:
"""
Add or update an object with a specific objectID.
Parameters:
- index_name: Target index name
- object_id: Unique object identifier
- body: Object data
- request_options: Additional request options
Returns:
UpdatedAtWithObjectIdResponse with task information
"""
async def save_objects_with_transformation(
self,
index_name: str,
objects: List[Dict[str, Any]],
wait_for_tasks: bool = False,
batch_size: int = 1000,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> List[IngestionWatchResponse]:
"""
Save objects with data transformation applied.
Parameters:
- index_name: Target index name
- objects: Objects to save with transformation
- wait_for_tasks: Wait for all tasks to complete
- batch_size: Objects per batch
- request_options: Additional request options
Returns:
List of IngestionWatchResponse objects
"""Index Management
Manage indices including creation, deletion, copying, and settings configuration.
async def list_indices(
self,
page: Optional[int] = None,
hits_per_page: Optional[int] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> ListIndicesResponse:
"""
List all indices in the application.
Parameters:
- page: Page number for pagination
- hits_per_page: Number of indices per page
- request_options: Additional request options
Returns:
ListIndicesResponse with index information
"""
async def delete_index(
self,
index_name: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> DeletedAtResponse:
"""
Delete an index and all its data.
Parameters:
- index_name: Index name to delete
- request_options: Additional request options
Returns:
DeletedAtResponse with task information
"""
async def operation_index(
self,
index_name: str,
operation_index_params: Union[OperationIndexParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Perform index operations (copy, move).
Parameters:
- index_name: Source index name
- operation_index_params: Operation parameters
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""
async def clear_objects(
self,
index_name: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Clear all objects from an index while preserving settings and rules.
Parameters:
- index_name: Target index name
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""
def index_exists(self, index_name: str) -> bool:
"""
Check if an index exists.
Parameters:
- index_name: Index name to check
Returns:
True if index exists, False otherwise
"""Settings and Configuration
Manage index settings including searchable attributes, ranking, facets, and other configuration options.
async def get_settings(
self,
index_name: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SettingsResponse:
"""
Get current index settings.
Parameters:
- index_name: Target index name
- request_options: Additional request options
Returns:
SettingsResponse with current settings
"""
async def set_settings(
self,
index_name: str,
index_settings: Union[IndexSettings, dict],
forward_to_replicas: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Update index settings.
Parameters:
- index_name: Target index name
- index_settings: New settings configuration
- forward_to_replicas: Apply to replica indices
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""Rules Management
Manage search rules for query processing, result ranking, and merchandising.
async def search_rules(
self,
index_name: str,
search_rules_params: Optional[Union[SearchRulesParams, dict]] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchRulesResponse:
"""
Search for rules in an index.
Parameters:
- index_name: Target index name
- search_rules_params: Search parameters
- request_options: Additional request options
Returns:
SearchRulesResponse with matching rules
"""
async def get_rule(
self,
index_name: str,
object_id: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> Rule:
"""
Get a specific rule by its objectID.
Parameters:
- index_name: Target index name
- object_id: Rule identifier
- request_options: Additional request options
Returns:
Rule object
"""
async def save_rule(
self,
index_name: str,
object_id: str,
rule: Union[Rule, dict],
forward_to_replicas: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Save a rule to an index.
Parameters:
- index_name: Target index name
- object_id: Rule identifier
- rule: Rule configuration
- forward_to_replicas: Apply to replica indices
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""
async def save_rules(
self,
index_name: str,
rules: Union[List[Rule], list[dict]],
forward_to_replicas: Optional[bool] = None,
clear_existing_rules: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Save multiple rules to an index.
Parameters:
- index_name: Target index name
- rules: List of rules to save
- forward_to_replicas: Apply to replica indices
- clear_existing_rules: Clear existing rules before saving
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""
async def delete_rule(
self,
index_name: str,
object_id: str,
forward_to_replicas: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Delete a specific rule from an index.
Parameters:
- index_name: Target index name
- object_id: Rule identifier to delete
- forward_to_replicas: Apply to replica indices
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""
async def clear_rules(
self,
index_name: str,
forward_to_replicas: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Clear all rules from an index.
Parameters:
- index_name: Target index name
- forward_to_replicas: Apply to replica indices
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""
async def browse_rules(
self,
index_name: str,
aggregator: Callable[[SearchRulesResponse], None],
search_rules_params: Optional[SearchRulesParams] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchRulesResponse:
"""
Browse all rules in an index with pagination.
Parameters:
- index_name: Target index name
- aggregator: Function to process each page of results
- search_rules_params: Search parameters
- request_options: Additional request options
Returns:
SearchRulesResponse with rules
"""API Key Management
Manage API keys for authentication and access control.
async def add_api_key(
self,
api_key: Union[ApiKey, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> AddApiKeyResponse:
"""
Create a new API key.
Parameters:
- api_key: API key configuration
- request_options: Additional request options
Returns:
AddApiKeyResponse with key information
"""
async def get_api_key(
self,
key: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetApiKeyResponse:
"""
Get API key details.
Parameters:
- key: API key value
- request_options: Additional request options
Returns:
GetApiKeyResponse with key details
"""
async def list_api_keys(
self,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> ListApiKeysResponse:
"""
List all API keys for the application.
Returns:
ListApiKeysResponse with all keys
"""
async def update_api_key(
self,
key: str,
api_key: Union[ApiKey, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdateApiKeyResponse:
"""
Update an existing API key.
Parameters:
- key: API key value to update
- api_key: Updated key configuration
- request_options: Additional request options
Returns:
UpdateApiKeyResponse with update confirmation
"""
async def delete_api_key(
self,
key: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> DeleteApiKeyResponse:
"""
Delete an API key.
Parameters:
- key: API key value to delete
- request_options: Additional request options
Returns:
DeleteApiKeyResponse with deletion confirmation
"""
async def generate_secured_api_key(
self,
parent_api_key: str,
restrictions: Union[SecuredApiKeyRestrictions, dict]
) -> str:
"""
Generate a secured API key with restrictions.
Parameters:
- parent_api_key: Base API key
- restrictions: Security restrictions to apply
Returns:
Secured API key string
"""
async def restore_api_key(
self,
key: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> AddApiKeyResponse:
"""
Restore a deleted API key.
Parameters:
- key: API key value to restore
- request_options: Additional request options
Returns:
AddApiKeyResponse with restoration confirmation
"""
async def wait_for_api_key(
self,
key: str,
operation: str,
api_key: Optional[Union[ApiKey, dict]] = None,
max_retries: int = 50,
timeout: RetryTimeout = RetryTimeout(),
request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetApiKeyResponse:
"""
Wait for an API key operation to be processed.
Parameters:
- key: API key value
- operation: Operation type ("add", "update", "delete")
- api_key: API key configuration for validation
- max_retries: Maximum retry attempts
- timeout: Timeout configuration
- request_options: Additional request options
Returns:
GetApiKeyResponse when operation completes
"""
def get_secured_api_key_remaining_validity(self, secured_api_key: str) -> int:
"""
Get remaining validity time for a secured API key.
Parameters:
- secured_api_key: Secured API key to check
Returns:
Remaining validity time in seconds
"""Task Management
Monitor and wait for asynchronous operations to complete.
async def wait_for_task(
self,
index_name: str,
task_id: int,
max_retries: int = 50,
timeout: RetryTimeout = RetryTimeout()
) -> GetTaskResponse:
"""
Wait for an indexing task to complete.
Parameters:
- index_name: Target index name
- task_id: Task identifier to wait for
- max_retries: Maximum retry attempts
- timeout: Timeout configuration
Returns:
GetTaskResponse when task completes
"""
async def get_task(
self,
index_name: str,
task_id: int,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetTaskResponse:
"""
Get the status of a specific task.
Parameters:
- index_name: Target index name
- task_id: Task identifier
- request_options: Additional request options
Returns:
GetTaskResponse with current task status
"""
async def get_app_task(
self,
task_id: int,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetTaskResponse:
"""
Get the status of an application-level task.
Parameters:
- task_id: Task identifier
- request_options: Additional request options
Returns:
GetTaskResponse with current task status
"""
async def wait_for_app_task(
self,
task_id: int,
timeout: RetryTimeout = RetryTimeout(),
max_retries: int = 50,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetTaskResponse:
"""
Wait for an application-level task to complete.
Parameters:
- task_id: Task identifier to wait for
- timeout: Timeout configuration
- max_retries: Maximum retry attempts
- request_options: Additional request options
Returns:
GetTaskResponse when task completes
"""Synonyms Management
Manage synonyms to improve search relevance by treating different terms as equivalent.
async def get_synonym(
self,
index_name: str,
object_id: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SynonymHit:
"""
Get a specific synonym by its objectID.
Parameters:
- index_name: Target index name
- object_id: Synonym identifier
- request_options: Additional request options
Returns:
SynonymHit object
"""
async def save_synonym(
self,
index_name: str,
object_id: str,
synonym_hit: Union[SynonymHit, dict],
forward_to_replicas: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SaveSynonymResponse:
"""
Save a synonym to an index.
Parameters:
- index_name: Target index name
- object_id: Synonym identifier
- synonym_hit: Synonym configuration
- forward_to_replicas: Apply to replica indices
- request_options: Additional request options
Returns:
SaveSynonymResponse with task information
"""
async def save_synonyms(
self,
index_name: str,
synonym_hit: Union[List[SynonymHit], list[dict]],
forward_to_replicas: Optional[bool] = None,
replace_existing_synonyms: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Save multiple synonyms to an index.
Parameters:
- index_name: Target index name
- synonym_hit: List of synonyms to save
- forward_to_replicas: Apply to replica indices
- replace_existing_synonyms: Replace all existing synonyms
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""
async def search_synonyms(
self,
index_name: str,
search_synonyms_params: Optional[Union[SearchSynonymsParams, dict]] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchSynonymsResponse:
"""
Search for synonyms in an index.
Parameters:
- index_name: Target index name
- search_synonyms_params: Search parameters
- request_options: Additional request options
Returns:
SearchSynonymsResponse with matching synonyms
"""
async def browse_synonyms(
self,
index_name: str,
aggregator: Callable[[SearchSynonymsResponse], None],
search_synonyms_params: Optional[SearchSynonymsParams] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchSynonymsResponse:
"""
Browse all synonyms in an index with pagination.
Parameters:
- index_name: Target index name
- aggregator: Function to process each page of results
- search_synonyms_params: Search parameters
- request_options: Additional request options
Returns:
SearchSynonymsResponse with synonyms
"""
async def delete_synonym(
self,
index_name: str,
object_id: str,
forward_to_replicas: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> DeletedAtResponse:
"""
Delete a specific synonym from an index.
Parameters:
- index_name: Target index name
- object_id: Synonym identifier to delete
- forward_to_replicas: Apply to replica indices
- request_options: Additional request options
Returns:
DeletedAtResponse with task information
"""
async def clear_synonyms(
self,
index_name: str,
forward_to_replicas: Optional[bool] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Clear all synonyms from an index.
Parameters:
- index_name: Target index name
- forward_to_replicas: Apply to replica indices
- request_options: Additional request options
Returns:
UpdatedAtResponse with task information
"""Batch Operations
Perform bulk operations efficiently with batching and chunking support.
async def batch(
self,
index_name: str,
batch_write_params: Union[BatchWriteParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> BatchResponse:
"""
Perform batch write operations on an index.
Parameters:
- index_name: Target index name
- batch_write_params: Batch operations to perform
- request_options: Additional request options
Returns:
BatchResponse with task information
"""
async def multiple_batch(
self,
batch_params: Union[BatchParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> MultipleBatchResponse:
"""
Perform batch operations across multiple indices.
Parameters:
- batch_params: Multi-index batch operations
- request_options: Additional request options
Returns:
MultipleBatchResponse with task information
"""
async def chunked_batch(
self,
index_name: str,
objects: List[Dict[str, Any]],
action: Action = Action.ADDOBJECT,
wait_for_tasks: bool = False,
batch_size: int = 1000,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> List[BatchResponse]:
"""
Perform chunked batch operations for large datasets.
Parameters:
- index_name: Target index name
- objects: Objects to process
- action: Batch action to perform
- wait_for_tasks: Wait for all tasks to complete
- batch_size: Objects per batch
- request_options: Additional request options
Returns:
List of BatchResponse objects for each chunk
"""
async def replace_all_objects(
self,
index_name: str,
objects: List[Dict[str, Any]],
batch_size: int = 1000,
scopes: List[str] = ["settings", "rules", "synonyms"],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> ReplaceAllObjectsResponse:
"""
Replace all objects in an index atomically.
Parameters:
- index_name: Target index name
- objects: New objects to replace with
- batch_size: Objects per batch
- scopes: Index components to preserve
- request_options: Additional request options
Returns:
ReplaceAllObjectsResponse with operation details
"""User ID Management
Manage user ID mappings for personalization and analytics.
async def assign_user_id(
self,
x_algolia_user_id: str,
assign_user_id_params: Union[AssignUserIdParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> CreatedAtResponse:
"""
Assign a user ID to a cluster.
Parameters:
- x_algolia_user_id: User identifier
- assign_user_id_params: Assignment parameters
- request_options: Additional request options
Returns:
CreatedAtResponse with assignment confirmation
"""
async def get_user_id(
self,
user_id: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UserId:
"""
Get user ID details and cluster assignment.
Parameters:
- user_id: User identifier to lookup
- request_options: Additional request options
Returns:
UserId object with user details
"""
async def list_user_ids(
self,
page: Optional[int] = None,
hits_per_page: Optional[int] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> ListUserIdsResponse:
"""
List all user IDs in the application.
Parameters:
- page: Page number for pagination
- hits_per_page: Number of results per page
- request_options: Additional request options
Returns:
ListUserIdsResponse with user ID list
"""
async def remove_user_id(
self,
user_id: str,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> RemoveUserIdResponse:
"""
Remove a user ID from the cluster.
Parameters:
- user_id: User identifier to remove
- request_options: Additional request options
Returns:
RemoveUserIdResponse with removal confirmation
"""
async def search_user_ids(
self,
search_user_ids_params: Union[SearchUserIdsParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchUserIdsResponse:
"""
Search for user IDs matching criteria.
Parameters:
- search_user_ids_params: Search parameters
- request_options: Additional request options
Returns:
SearchUserIdsResponse with matching user IDs
"""
async def get_top_user_ids(
self,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetTopUserIdsResponse:
"""
Get the most active user IDs.
Parameters:
- request_options: Additional request options
Returns:
GetTopUserIdsResponse with top user IDs
"""Dictionary Operations
Manage custom dictionaries for enhanced search functionality.
async def get_dictionary_languages(
self,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> Dict[str, Languages]:
"""
Get available dictionary languages.
Parameters:
- request_options: Additional request options
Returns:
Dictionary of available languages
"""
async def get_dictionary_settings(
self,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> GetDictionarySettingsResponse:
"""
Get current dictionary settings.
Parameters:
- request_options: Additional request options
Returns:
GetDictionarySettingsResponse with settings
"""
async def set_dictionary_settings(
self,
dictionary_settings_params: Union[DictionarySettingsParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Update dictionary settings.
Parameters:
- dictionary_settings_params: New dictionary settings
- request_options: Additional request options
Returns:
UpdatedAtResponse with update confirmation
"""
async def search_dictionary_entries(
self,
dictionary_name: Union[DictionaryType, str],
search_dictionary_entries_params: Union[SearchDictionaryEntriesParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> SearchDictionaryEntriesResponse:
"""
Search for entries in a dictionary.
Parameters:
- dictionary_name: Dictionary type to search
- search_dictionary_entries_params: Search parameters
- request_options: Additional request options
Returns:
SearchDictionaryEntriesResponse with matching entries
"""
async def batch_dictionary_entries(
self,
dictionary_name: Union[DictionaryType, str],
batch_dictionary_entries_params: Union[BatchDictionaryEntriesParams, dict],
request_options: Optional[Union[dict, RequestOptions]] = None
) -> UpdatedAtResponse:
"""
Perform batch operations on dictionary entries.
Parameters:
- dictionary_name: Dictionary type to modify
- batch_dictionary_entries_params: Batch operations to perform
- request_options: Additional request options
Returns:
UpdatedAtResponse with operation confirmation
"""Custom Operations
Perform custom HTTP operations on the Algolia API.
async def custom_get(
self,
path: str,
parameters: Optional[Dict[str, Any]] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> object:
"""
Perform a custom GET request.
Parameters:
- path: API endpoint path after "/1"
- parameters: Query parameters
- request_options: Additional request options
Returns:
Response object
"""
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
) -> object:
"""
Perform a custom POST request.
Parameters:
- path: API endpoint path after "/1"
- parameters: Query parameters
- body: Request body data
- request_options: Additional request options
Returns:
Response object
"""
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
) -> object:
"""
Perform a custom PUT request.
Parameters:
- path: API endpoint path after "/1"
- parameters: Query parameters
- body: Request body data
- request_options: Additional request options
Returns:
Response object
"""
async def custom_delete(
self,
path: str,
parameters: Optional[Dict[str, Any]] = None,
request_options: Optional[Union[dict, RequestOptions]] = None
) -> object:
"""
Perform a custom DELETE request.
Parameters:
- path: API endpoint path after "/1"
- parameters: Query parameters
- request_options: Additional request options
Returns:
Response object
"""
## Usage Examples
### Basic Search and Object Management
```python
from algoliasearch.search.client import SearchClient
# Initialize client
client = SearchClient("YOUR_APP_ID", "YOUR_API_KEY")
# Save objects to an index
products = [
{"objectID": "1", "name": "iPhone 13", "brand": "Apple", "price": 699},
{"objectID": "2", "name": "Galaxy S21", "brand": "Samsung", "price": 649}
]
# Save objects with automatic chunking
responses = await client.save_objects(
index_name="products",
objects=products,
wait_for_tasks=True
)
# Search for products
search_response = await client.search({
"requests": [{
"indexName": "products",
"query": "iPhone",
"hitsPerPage": 10,
"filters": "price < 800"
}]
})
# Display results
for hit in search_response.results[0].hits:
print(f"{hit['name']} - ${hit['price']}")Advanced Search with Rules and Synonyms
# Create a search rule for merchandising
rule = {
"objectID": "promo-rule",
"condition": {
"pattern": "smartphone",
"anchoring": "contains"
},
"consequence": {
"promote": [
{"objectID": "1", "position": 0}
]
},
"description": "Promote iPhone for smartphone queries"
}
# Save the rule
await client.save_rule(
index_name="products",
object_id="promo-rule",
rule=rule
)
# Create synonyms for better search
synonym = {
"objectID": "phone-synonyms",
"type": "synonym",
"synonyms": ["phone", "mobile", "smartphone", "cellphone"]
}
# Save the synonym
await client.save_synonym(
index_name="products",
object_id="phone-synonyms",
synonym_hit=synonym
)Batch Operations and Index Management
# Replace all objects atomically
new_products = [
{"objectID": "3", "name": "iPad Pro", "brand": "Apple", "price": 1099},
{"objectID": "4", "name": "Surface Pro", "brand": "Microsoft", "price": 999}
]
# This preserves settings, rules, and synonyms while replacing all objects
replace_response = await client.replace_all_objects(
index_name="products",
objects=new_products,
batch_size=1000
)
# Configure index settings
settings = {
"searchableAttributes": ["name", "brand", "description"],
"attributesForFaceting": ["brand", "category", "price"],
"customRanking": ["desc(popularity)", "asc(price)"],
"ranking": ["typo", "geo", "words", "filters", "proximity", "attribute", "exact", "custom"]
}
await client.set_settings(
index_name="products",
index_settings=settings
)Types
# Core request/response types
class SearchMethodParams(BaseModel):
requests: List[SearchParams]
class SearchParams(BaseModel):
index_name: str
query: Optional[str] = None
hits_per_page: Optional[int] = None
page: Optional[int] = None
filters: Optional[str] = None
facet_filters: Optional[List[str]] = None
numeric_filters: Optional[List[str]] = None
around_lat_lng: Optional[str] = None
around_radius: Optional[int] = None
class SearchResponse(BaseModel):
results: List[SearchResult]
class SearchResult(BaseModel):
hits: List[dict]
nb_hits: int
page: int
nb_pages: int
hits_per_page: int
processing_time_ms: int
exhaustive_nb_hits: bool
query: str
params: str
# Index settings
class IndexSettings(BaseModel):
searchable_attributes: Optional[List[str]] = None
attributes_for_faceting: Optional[List[str]] = None
unretrievable_attributes: Optional[List[str]] = None
attributes_to_retrieve: Optional[List[str]] = None
ranking: Optional[List[str]] = None
custom_ranking: Optional[List[str]] = None
replicas: Optional[List[str]] = None
max_values_per_facet: Optional[int] = None
sort_facet_values_by: Optional[str] = None
# API Key types
class ApiKey(BaseModel):
acl: List[str]
description: Optional[str] = None
indices: Optional[List[str]] = None
max_hits_per_query: Optional[int] = None
max_queries_per_ip_per_hour: Optional[int] = None
query_parameters: Optional[str] = None
referers: Optional[List[str]] = None
validity: Optional[int] = None
class SecuredApiKeyRestrictions(BaseModel):
valid_until: Optional[int] = None
restrict_indices: Optional[List[str]] = None
restrict_sources: Optional[str] = None
filter: Optional[str] = None
user_token: Optional[str] = None
# Rules
class Rule(BaseModel):
object_id: str
condition: dict
consequence: dict
description: Optional[str] = None
enabled: Optional[bool] = None
# Synonyms
class SynonymHit(BaseModel):
object_id: str
type: str
synonyms: Optional[List[str]] = None
input: Optional[str] = None
word: Optional[str] = None
corrections: Optional[List[str]] = None
# Batch operations
class BatchWriteParams(BaseModel):
requests: List[dict]
class BatchParams(BaseModel):
requests: List[dict]
class Action(str, Enum):
ADDOBJECT = "addObject"
UPDATEOBJECT = "updateObject"
PARTIALUPDATEOBJECT = "partialUpdateObject"
DELETEOBJECT = "deleteObject"
# User ID types
class AssignUserIdParams(BaseModel):
cluster: str
class UserId(BaseModel):
user_id: str
cluster_name: str
nb_records: int
data_size: int
# Dictionary types
class DictionaryType(str, Enum):
STOPWORDS = "stopwords"
PLURALS = "plurals"
COMPOUNDS = "compounds"
class Languages(BaseModel):
nb_custom_entries: int
# Response types
class SaveObjectResponse(BaseModel):
created_at: str
task_id: int
object_id: str
class SaveSynonymResponse(BaseModel):
updated_at: str
task_id: int
id: str
class MultipleBatchResponse(BaseModel):
task_id: Dict[str, int]
object_ids: List[str]
class ReplaceAllObjectsResponse(BaseModel):
copy_operation_response: UpdatedAtResponse
batch_responses: List[BatchResponse]
move_operation_response: UpdatedAtResponse
class IngestionWatchResponse(BaseModel):
task_id: int
watched_objects: List[dict]
class DeleteByParams(BaseModel):
filters: Optional[str] = None
numeric_filters: Optional[List[str]] = None
facet_filters: Optional[List[str]] = None
around_lat_lng: Optional[str] = None
around_radius: Optional[int] = None
class SearchDictionaryEntriesParams(BaseModel):
query: Optional[str] = None
page: Optional[int] = None
hits_per_page: Optional[int] = None
class BatchDictionaryEntriesParams(BaseModel):
clear_existing_dictionary_entries: Optional[bool] = None
requests: List[dict]
class SearchUserIdsParams(BaseModel):
query: Optional[str] = None
cluster_name: Optional[str] = None
page: Optional[int] = None
hits_per_page: Optional[int] = NoneInstall with Tessl CLI
npx tessl i tessl/pypi-algoliasearch