API clients / Python

Helper methods

Helper methods extend or combine API operations to make using the Search API easier.

Browse for records

Required ACL: browse

Retrieves all records from an index.

This helper method iterates over the paginated API response from the Browse API operation and lets you run an aggregator function on every iteration.

You can use this, for example, to retrieve all records from your index.

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13

def browse_objects(
    index_name: str,
    aggregator: Callable[[BrowseResponse], None],
    browse_params: BrowseParams,
    request_options: dict | RequestOptions | None,
): BrowseResponse

async def browse_objects(
    index_name: str,
    aggregator: Callable[[BrowseResponse], None],
    browse_params: BrowseParams,
    request_options: dict | RequestOptions | None,
): BrowseResponse

Parameters

index_name
type: str
Required

Index name from which to get all records.
aggregator
type: Callable[[BrowseResponse], None]
Required

A function that runs on every iteration, for example, to aggregate all records from the response.
browse_params
type: BrowseParams
default: BrowseParams()

Parameters for the Browse for records request.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.

Browse for rules

Required ACL: settings

Retrieves all rules from an index.

This helper method iterates over the paginated API response from the Browse API operation and lets you run an aggregator function on every iteration. You can use this, for example, to retrieve all records from your index.

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13

def browse_rules(
    index_name: str,
    aggregator: Callable[[SearchRulesResponse], None],
    search_rules_params: SearchRulesParams,
    request_options: dict | RequestOptions | None,
): SearchRulesResponse

async def browse_rules(
    index_name: str,
    aggregator: Callable[[SearchRulesResponse], None],
    search_rules_params: SearchRulesParams,
    request_options: dict | RequestOptions | None,
): SearchRulesResponse

Parameters

index_name
type: str
Required

Index name from which to get all rules.
aggregator
type: Callable[[SearchRulesResponse], None]
Required

A function that runs on every iteration, for example, to aggregate all rules from the response.
search_rules_params
type: SearchRulesParams
default: SearchRulesParams(hits_per_page=1000)

Parameters for the Search for rules request.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.

Browse for synonyms

Required ACL: settings

Retrieves all synonyms from an index.

This helper method iterates over the paginated API response from the Search for synonyms API operation and lets you run an aggregator function on every iteration.

You can use this, for example, to retrieve all synonyms from your index.

Copy 
1
2
3
4
5
6

client.browse_synonyms(
    index_name: str,
    aggregator: Callable[[SearchSynonymsResponse], None],
    search_synonyms_params: SearchSynonymsParams,
    request_options: dict | RequestOptions | None,
)

Parameters

index_name
type: str
Required

Index name from which to get all synonyms.
aggregator
type: Callable[[SearchSynonymsResponse], None]
Required

A function that runs on every iteration, for example, to aggregate all synonyms from the response.
search_synonyms_params
type: SearchSynonymsParams
default: SearchSynonymsParams(hits_per_page=1000)

Parameters for the Search for synonyms request.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.

Delete records

Required ACL: deleteObject

Deletes records from an index using their object IDs.

This helper method creates batch write requests with deleteObject actions and sends these requests in batches of 1,000.

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13

def delete_objects(
    index_name: str,
    object_ids: list[str],
    wait_for_tasks: bool,
    request_options: dict | RequestOptions | None,
): list[BatchResponse]

async def delete_objects(
    index_name: str,
    object_ids: list[str],
    wait_for_tasks: bool,
    request_options: dict | RequestOptions | None,
): list[BatchResponse]

Examples

Copy 
1
2
3
4
5
6
7
8
9
10
11
12

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.delete_objects(
    index_name="ALGOLIA_INDEX_NAME",
    object_ids=[
        "1",
        "2",
    ],
)

Parameters

index_name
type: str
Required

Index name from which to delete records.
object_ids
type: list
items: str
Required

Object IDs to delete.
wait_for_tasks
type: bool
default: False

Whether to wait until all tasks created by this method are completed.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.

Create secured API keys

Creates a secured API key without any calls to Algolia’s servers.

This helper method takes a parent API key and adds restrictions to it, which can’t be changed at query time.

Secured API keys are forms of hash-based message authentication codes (HMAC).

This helper performs these steps to create a secured API key:

  1. Compute a SHA-256 HMAC with the parent API key as secret, and a URL-encoded list of query parameters as message. These query parameters will be applied with every use of the API key and can’t be changed by users.

  2. Concatenate the SHA-256 HMAC with the list of query parameters to a single string.

  3. Encode the whole string in base64.

Copy 
1
2
3
4
5
6
7
8
9

def generate_secured_api_key(
    parent_api_key: str,
    restrictions: dict | SecuredApiKeyRestrictions | None,
): str

async def generate_secured_api_key(
    parent_api_key: str,
    restrictions: dict | SecuredApiKeyRestrictions | None,
): str

Examples

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.generate_secured_api_key(
    parent_api_key="2640659426d5107b6e47d75db9cbaef8",
    restrictions={
        "validUntil": 2524604400,
        "restrictIndices": [
            "Movies",
            "cts_e2e_settings",
        ],
        "restrictSources": "192.168.1.0/24",
        "filters": "category:Book OR category:Ebook AND _tags:published",
        "userToken": "user123",
        "searchParams": {
            "query": "batman",
            "typoTolerance": "strict",
            "aroundRadius": "all",
            "mode": "neuralSearch",
            "hitsPerPage": 10,
            "optionalWords": [
                "one",
                "two",
            ],
        },
    },
)

Parameters

parent_api_key
type: str
Required

The API key to be used as secret for the secured API key. The secured API key inherits all restrictions from this API key. You can’t use an Admin API key.
restrictions
type: dict | SecuredApiKeyRestrictions | None
default: SecuredApiKeyRestrictions()

Restrictions to apply to the secured API key. You must add at least one restriction. If you make requests with a secured API key with the same restrictions as the parent_api_key, the API returns an error with the status 403.

restrictions ➔ SecuredApiKeyRestrictions

Restrictions you can add to secured API keys.

search_params
type: SearchParams | None
default: None

Search parameters that will always be applied with this API key. For more information, see Search.
filters
type: str | None
default: None

Filters that will always be applied with this API key.
valid_until
type: int | None
default: None

Timestamp when the API key expires, in seconds since the Unix epoch.
restrict_indices
type: list[str] | None
default: None

Index names that this API key should be allowed to access. You can use the wildcard character *. For example, restrict_indices: "dev_*" permits using all indices with names starting with dev_.
restrict_sources
type: str | None
default: None

IP network range that is allowed to use this API key.
user_token
type: str | None
default: None

Pseudonymous user identifier to restrict usage of this API key to specific users. By default, rate limits are set based on IP addresses. This can be an issue if many users search from the same IP address. To avoid this, add a user token to each generated API key.

Get secured API key remaining validity

Retrieves the time in seconds until the secured API key expires.

This helper method returns the amount of seconds left until the secured API key expires, as defined by its validUntil property. If the key already expired, this method returns a negative number, indicating the number of seconds since the key expired.

Copy 
1
2
3

def client.get_secured_api_key_remaining_validity(
  secured_api_key: str
): int

Parameters

secured_api_key
type: str
Required

Secured API key for which to retrieve the time of validity.

Check if index exists

Required ACL: settings

Checks whether an index exists.

This helper method makes a request to the Retrieve index settings endpoint, and returns true if that request returns a success response.

Copy 
1
2
3
4
5
6
7

def index_exists(
  index_name: str,
): bool

async def index_exists(
  index_name: str,
): bool

Parameters

index_name
type: str
Required

Index name to check.

Update attributes of records

Required ACL: addObject

Adds or updates attributes of records

This helper method creates batch write requests with partialUpdateObject or partialUpdateObjectNoCreate actions (depending on whether the create_if_not_exists option is true or false) and sends these requests in batches of 1,000.

The underlying method is subject to indexing rate limits.

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

def partial_update_objects(
    index_name: str,
    objects: list[dict[str, Any]],
    create_if_not_exists: bool,
    wait_for_tasks: bool,
    request_options: dict | RequestOptions | None,
): list[BatchResponse]

async def partial_update_objects(
    index_name: str,
    objects: list[dict[str, Any]],
    create_if_not_exists: bool,
    wait_for_tasks: bool,
    request_options: dict | RequestOptions | None,
): list[BatchResponse]

Examples

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.partial_update_objects(
    index_name="ALGOLIA_INDEX_NAME",
    objects=[
        {
            "objectID": "1",
            "name": "Adam",
        },
        {
            "objectID": "2",
            "name": "Benoit",
        },
    ],
    create_if_not_exists=True,
)

Parameters

index_name
type: str
Required

Index name where to update records.
objects
type: list
items: dict[str, Any]
Required

Records to update
create_if_not_exists
type: bool
default: False

Whether to create a new record if it doesn’t exist.
wait_for_tasks
type: bool
default: False

Whether to wait until all tasks created by this method are completed.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.

Replace all records

Required ACL: addObject

Replaces all records in an index with new ones.

This helper lets you replace all records in an index without interrupting the ability to search.

It combines batch, copy, and move requests:

  1. Copy settings, synonyms, and rules to a temporary index.
  2. Add the records from the objects parameter to the temporary index.
  3. Move the temporary index to the original index to replace it.

If you use an API key with indexes restrictions, you need to allow access to the index INDEX_NAME_tmp_* (replace INDEX_NAME with the name of your original index).

The underlying methods are subject to indexing rate limits.

The response from this method contains the API responses from the individual requests:

Copy 
1
2
3
4

class ReplaceAllObjectsResponse:
  copy_operation_response: UpdatedAtResponse
  batch_responses: List[BatchResponse]
  move_operation_response: UpdatedAtResponse
Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13

def replace_all_objects(
    index_name: str,
    objects: list[dict[str, Any]],
    batch_size: int,
    request_options: dict | RequestOptions | None,
): ReplaceAllObjectsResponse

async def replace_all_objects(
    index_name: str,
    objects: list[dict[str, Any]],
    batch_size: int,
    request_options: dict | RequestOptions | None,
): ReplaceAllObjectsResponse

Examples

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.replace_all_objects(
    index_name="ALGOLIA_INDEX_NAME",
    objects=[
        {
            "objectID": "1",
            "name": "Adam",
        },
        {
            "objectID": "2",
            "name": "Benoit",
        },
        {
            "objectID": "3",
            "name": "Cyril",
        },
        {
            "objectID": "4",
            "name": "David",
        },
        {
            "objectID": "5",
            "name": "Eva",
        },
        {
            "objectID": "6",
            "name": "Fiona",
        },
        {
            "objectID": "7",
            "name": "Gael",
        },
        {
            "objectID": "8",
            "name": "Hugo",
        },
        {
            "objectID": "9",
            "name": "Igor",
        },
        {
            "objectID": "10",
            "name": "Julia",
        },
    ],
    batch_size=3,
)

Parameters

index_name
type: str
Required

Index name where to replace all records.
objects
type: list
items: dict[str,Any]
Required

Records that replace the existing records in your index.
batch_size
type: int
default: 1,000

Batch size for the indexing operation.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.

Save records

Required ACL: addObject

Adds records to an index.

This helper method creates batch write requests with addObject actions and sends these requests in batches in 1,000.

The underlying method is subject to indexing rate limits.

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13

def save_objects(
    index_name: str,
    objects: list[dict[str,Any]],
    wait_for_tasks: bool,
    request_options: dict | RequestOptions | None,
): list[BatchResponse]

async def save_objects(
    index_name: str,
    objects: list[dict[str,Any]],
    wait_for_tasks: bool,
    request_options: dict | RequestOptions | None,
): list[BatchResponse]

Examples

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.save_objects(
    index_name="ALGOLIA_INDEX_NAME",
    objects=[
        {
            "objectID": "1",
            "name": "Adam",
        },
        {
            "objectID": "2",
            "name": "Benoit",
        },
    ],
)

Parameters

index_name
type: str
Required

Index name to which to add records.
objects
type: list
items: dict[str, Any]
Required

Records to add.
wait_for_tasks
type: bool
default: False

Whether to wait until all tasks created by this method are completed.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.

Wait for API keys

Waits for an API key operation to be finished.

When you add, update, or delete API keys, it might take a while before the change is applied.

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

def wait_for_api_key(
    key: str,
    operation: str,
    api_key: dict[str,Any] | ApiKey | None,
    max_retries: int,
    timeout: Callable[[int], int],
    request_options: dict | RequestOptions | None,
): GetApiKeyResponse | None

async def wait_for_api_key(
    key: str,
    operation: str,
    api_key: dict[str,Any] | ApiKey | None,
    max_retries: int,
    timeout: Callable[[int], int],
    request_options: dict | RequestOptions | None,
): GetApiKeyResponse | None

Examples

Wait until an API key is added:

Copy 
1
2
3
4
5
6
7
8
9

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.wait_for_api_key(
    key="api-key-add-operation-test-python",
    operation="add",
)

Wait until an API key is updated:

Copy 
1
2
3
4
5
6
7
8
9

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.wait_for_api_key(
    key="api-key-add-operation-test-python",
    operation="add",
)

Wait until an API key is deleted:

Copy 
1
2
3
4
5
6
7
8
9

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.wait_for_api_key(
    key="api-key-add-operation-test-python",
    operation="add",
)

Parameters

key
type: str
Required

The API key that was added, updated, or deleted. For example, you can get this value from the API response when creating or updating.
operation
type: str
Required

The type of operation to wait for:

  • add. Wait for the addition of a new API key
  • update. Wait for the update of an existing API key
  • delete. Wait for the deletion of an API key.
api_key
type: dict[str, Any] | ApiKey | None
default: None

When waiting for an API key to be updated, api_key expects the original ApiKey object you passed to the Update API key operation.
max_retries
type: int
default: 50

Maximum number of times the API key’s status should be checked.
timeout
type: Callable[[int], int]
default: min(retry_count*0.2, 5)

A function that takes one parameter, the current number of retries, and returns the number of seconds after which to check the API key’s status again.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.

Wait for application-level tasks

Waits for an application-level task to complete.

While indexing operations are index-level tasks, some tasks, such as working with dictionaries, are application-level tasks. They’re asynchronous: when you make a request, Algolia adds the task to a queue and runs it depending on the server load. The response includes a task_id property that you can use to wait until the task is complete.

This helper method polls the Check application task status API operation and stops if the task’s status is published.

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13

def wait_for_app_task(
    task_id: str,
    max_retries: int,
    timeout: Callable[[int], int],
    request_options: dict | RequestOptions | None,
): GetTaskResponse

async def client.wait_for_app_task(
    task_id: str,
    max_retries: int,
    timeout: Callable[[int], int],
    request_options: dict | RequestOptions | None,
): GetTaskResponse

Examples

Copy 
1
2
3
4
5
6
7
8

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.wait_for_app_task(
    task_id=123,
)

Parameters

task_id
type: str
Required

Task ID for which to wait.
max_retries
type: int
default: 50

Maximum number of times the task status should be checked.
timeout
type: Callable[[int], int]
default: min(retry_count*0.2, 5)

A function that takes one parameter, the current number of retries, and returns the number of seconds after which to check the task’s status again.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.

Wait for index operations

Waits for an indexing task to complete.

Indexing operations are asynchronous. When you make a request for an indexing operation, such as, adding or updating records, Algolia adds a task to a queue and returns a task_id property with the response. The task itself runs separately, depending on the server load.

This helper method polls the Check task status API operation and stops if the task’s status is published.

Copy 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

def wait_for_task(
    index_name: str,
    task_id: str,
    max_retries: int,
    timeout: Callable[[int], int],
    request_options: dict | RequestOptions | None,
): GetTaskResponse

async def wait_for_task(
    index_name: str,
    task_id: str,
    max_retries: int,
    timeout: Callable[[int], int],
    request_options: dict | RequestOptions | None,
): GetTaskResponse

Examples

Copy 
1
2
3
4
5
6
7
8
9

from algoliasearch.search.client import SearchClientSync

# In an asynchronous context, you can use SearchClient instead, which exposes the exact same methods.
client = SearchClientSync("ALGOLIA_APPLICATION_ID", "ALGOLIA_API_KEY")

response = client.wait_for_task(
    index_name="ALGOLIA_INDEX_NAME",
    task_id=123,
)

Parameters

index_name
type: str
Required

Index name for which to check the task.
task_id
type: str
Required

Task ID for which to wait.
max_retries
type: int
default: 50

Maximum number of times the task status should be checked.
timeout
type: Callable[[int], int]
default: min(retry_count*0.2, 5)

A function that takes one parameter, the current number of retries, and returns the number of seconds after which to check the task’s status again.
request_options
type: dict | RequestOptions | None
default: None

Extra parameters, such as headers or query parameters. For more information, see Request options.
Did you find this page helpful?