AsyncClient

AsyncClient Usage

Create a client with a context manager

This client runs in a context manager which ensures that everything is cleaned up after the use of the client is done. To create a client:

from meilisearch-python-sdk import AsyncClient


async with AsyncClient("http://localhost:7700", "masterKey") as client:
    index = client.index("movies")
    ...

Custom headers

Custom headers can be added to the client by adding them to custom_headers when creating the client.

from meilisearch_python_sdk import AsyncClient

async with AsyncClient(
    "http://127.0.0.1:7700",
    "masterKey",
    custom_headers={"header_key_1": "header_value_1", "header_key_2": "header_value_2"}
) as client:
    index = client.index("movies")
    ...

Create a client without a context manager

It is also possible to call the client without using a context manager, but in doing so you will need to make sure to do the cleanup yourself:

from meilisearch-python-sdk import AsyncClient


try:
    client = AsyncClient("http://localhost:7700", "masterKey")
    ...
finally:
    await client.aclose()

AsyncClient API

Bases: BaseClient

Async client to connect to the Meilisearch API.

Source code in meilisearch_python_sdk/_client.py

class AsyncClient(BaseClient):
    """Async client to connect to the Meilisearch API."""

    def __init__(
        self,
        url: str,
        api_key: str | None = None,
        *,
        timeout: int | None = None,
        verify: str | bool | SSLContext = True,
        custom_headers: dict[str, str] | None = None,
        json_handler: BuiltinHandler | OrjsonHandler | UjsonHandler | None = None,
        http2: bool = False,
    ) -> None:
        """Class initializer.

        Args:
            url: The url to the Meilisearch API (ex: http://localhost:7700)
            api_key: The optional API key for Meilisearch. Defaults to None.
            timeout: The amount of time in seconds that the client will wait for a response before
                timing out. Defaults to None.
            verify: SSL certificates (a.k.a CA bundle) used to
                verify the identity of requested hosts. Either `True` (default CA bundle),
                a path to an SSL certificate file, or `False` (disable verification)
            custom_headers: Custom headers to add when sending data to Meilisearch. Defaults to
                None.
            json_handler: The module to use for json operations. The options are BuiltinHandler
                (uses the json module from the standard library), OrjsonHandler (uses orjson), or
                UjsonHandler (uses ujson). Note that in order use orjson or ujson the corresponding
                extra needs to be included. Default: BuiltinHandler.
            http2: Whether or not to use HTTP/2. Defaults to False.
        """
        super().__init__(api_key, custom_headers, json_handler)

        self.http_client = HttpxAsyncClient(
            base_url=url, timeout=timeout, headers=self._headers, verify=verify, http2=http2
        )
        self._http_requests = AsyncHttpRequests(self.http_client, json_handler=self.json_handler)

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(
        self,
        et: type[BaseException] | None,
        ev: type[BaseException] | None,
        traceback: TracebackType | None,
    ) -> None:
        await self.aclose()

    async def aclose(self) -> None:
        """Closes the client.

        This only needs to be used if the client was not created with a context manager.
        """
        await self.http_client.aclose()

    async def create_dump(self) -> TaskInfo:
        """Trigger the creation of a Meilisearch dump.

        Returns:
            The details of the task.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples:
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     await client.create_dump()
        """
        response = await self._http_requests.post("dumps")

        return TaskInfo(**response.json())

    async def create_index(
        self,
        uid: str,
        primary_key: str | None = None,
        *,
        settings: MeilisearchSettings | None = None,
        wait: bool = True,
        timeout_in_ms: int | None = None,
        plugins: AsyncIndexPlugins | None = None,
        hits_type: Any = JsonDict,
    ) -> AsyncIndex:
        """Creates a new index.

        Args:
            uid: The index's unique identifier.
            primary_key: The primary key of the documents. Defaults to None.
            settings: Settings for the index. The settings can also be updated independently of
                creating the index. The advantage to updating them here is updating the settings after
                adding documents will cause the documents to be re-indexed. Because of this it will be
                faster to update them before adding documents. Defaults to None (i.e. default
                Meilisearch index settings).
            wait: If set to True and settings are being updated, the index will be returned after
                the settings update has completed. If False it will not wait for settings to complete.
                Default: True
            timeout_in_ms: Amount of time in milliseconds to wait before raising a
                MeilisearchTimeoutError. `None` can also be passed to wait indefinitely. Be aware that
                if the `None` option is used the wait time could be very long. Defaults to None.
            plugins: Optional plugins can be provided to extend functionality.
            hits_type: Allows for a custom type to be passed to use for hits. Defaults to
                JsonDict

        Returns:
            An instance of AsyncIndex containing the information of the newly created index.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples:
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     index = await client.create_index("movies")
        """
        return await AsyncIndex.create(
            self.http_client,
            uid,
            primary_key,
            settings=settings,
            wait=wait,
            timeout_in_ms=timeout_in_ms,
            plugins=plugins,
            json_handler=self.json_handler,
            hits_type=hits_type,
        )

    async def create_snapshot(self) -> TaskInfo:
        """Trigger the creation of a Meilisearch snapshot.

        Returns:
            The details of the task.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples:
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     await client.create_snapshot()
        """
        response = await self._http_requests.post("snapshots")

        return TaskInfo(**response.json())

    async def delete_index_if_exists(self, uid: str) -> bool:
        """Deletes an index if it already exists.

        Args:
            uid: The index's unique identifier.

        Returns:
            True if an index was deleted for False if not.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples:
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     await client.delete_index_if_exists()
        """
        response = await self._http_requests.delete(f"indexes/{uid}")
        status = await self.wait_for_task(response.json()["taskUid"], timeout_in_ms=100000)
        if status.status == "succeeded":
            return True
        return False

    async def get_indexes(
        self, *, offset: int | None = None, limit: int | None = None
    ) -> list[AsyncIndex] | None:
        """Get all indexes.

        Args:
            offset: Number of indexes to skip. The default of None will use the Meilisearch
                default.
            limit: Number of indexes to return. The default of None will use the Meilisearch
                default.

        Returns:
            A list of all indexes.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples:
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     indexes = await client.get_indexes()
        """
        url = _build_offset_limit_url("indexes", offset, limit)
        response = await self._http_requests.get(url)

        if not response.json()["results"]:
            return None

        return [
            AsyncIndex(
                http_client=self.http_client,
                uid=x["uid"],
                primary_key=x["primaryKey"],
                created_at=x["createdAt"],
                updated_at=x["updatedAt"],
                json_handler=self.json_handler,
            )
            for x in response.json()["results"]
        ]

    async def get_index(self, uid: str) -> AsyncIndex:
        """Gets a single index based on the uid of the index.

        Args:
            uid: The index's unique identifier.

        Returns:
            An AsyncIndex instance containing the information of the fetched index.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples:
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     index = await client.get_index()
        """
        return await AsyncIndex(self.http_client, uid, json_handler=self.json_handler).fetch_info()

    def index(self, uid: str, *, plugins: AsyncIndexPlugins | None = None) -> AsyncIndex:
        """Create a local reference to an index identified by UID, without making an HTTP call.

        Because no network call is made this method is not awaitable.

        Args:
            uid: The index's unique identifier.
            plugins: Optional plugins can be provided to extend functionality.

        Returns:
            An AsyncIndex instance.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples:
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     index = client.index("movies")
        """
        return AsyncIndex(
            self.http_client, uid=uid, plugins=plugins, json_handler=self.json_handler
        )

    async def get_all_stats(self) -> ClientStats:
        """Get stats for all indexes.

        Returns:
            Information about database size and all indexes.
            https://docs.meilisearch.com/reference/api/stats.html

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     stats = await client.get_all_stats()
        """
        response = await self._http_requests.get("stats")

        return ClientStats(**response.json())

    async def get_or_create_index(
        self,
        uid: str,
        primary_key: str | None = None,
        *,
        plugins: AsyncIndexPlugins | None = None,
        hits_type: Any = JsonDict,
    ) -> AsyncIndex:
        """Get an index, or create it if it doesn't exist.

        Args:
            uid: The index's unique identifier.
            primary_key: The primary key of the documents. Defaults to None.
            plugins: Optional plugins can be provided to extend functionality.
            hits_type: Allows for a custom type to be passed to use for hits. Defaults to
                JsonDict

        Returns:
            An instance of AsyncIndex containing the information of the retrieved or newly created index.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.MeilisearchTimeoutError: If the connection times out.
            MeilisearchTimeoutError: If the connection times out.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     index = await client.get_or_create_index("movies")
        """
        try:
            index_instance = await self.get_index(uid)
        except MeilisearchApiError as err:
            if "index_not_found" not in err.code:
                raise
            index_instance = await self.create_index(
                uid, primary_key, plugins=plugins, hits_type=hits_type
            )
        return index_instance

    async def create_key(self, key: KeyCreate) -> Key:
        """Creates a new API key.

        Args:
            key: The information to use in creating the key. Note that if an expires_at value
                is included it should be in UTC time.

        Returns:
            The new API key.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> from meilissearch_async_client.models.client import KeyCreate
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     key_info = KeyCreate(
            >>>         description="My new key",
            >>>         actions=["search"],
            >>>         indexes=["movies"],
            >>>     )
            >>>     keys = await client.create_key(key_info)
        """
        response = await self._http_requests.post(
            "keys", self.json_handler.loads(key.model_dump_json(by_alias=True))
        )  # type: ignore[attr-defined]

        return Key(**response.json())

    async def delete_key(self, key: str) -> int:
        """Deletes an API key.

        Args:
            key: The key or uid to delete.

        Returns:
            The Response status code. 204 signifies a successful delete.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     await client.delete_key("abc123")
        """
        response = await self._http_requests.delete(f"keys/{key}")
        return response.status_code

    async def get_keys(self, *, offset: int | None = None, limit: int | None = None) -> KeySearch:
        """Gets the Meilisearch API keys.

        Args:
            offset: Number of indexes to skip. The default of None will use the Meilisearch
                default.
            limit: Number of indexes to return. The default of None will use the Meilisearch
                default.

        Returns:
            API keys.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            from meilisearch_python_sdk import AsyncClient
            async with AsyncClient("http://localhost.com", "masterKey") as client:
                keys = await client.get_keys()
        """
        url = _build_offset_limit_url("keys", offset, limit)
        response = await self._http_requests.get(url)

        return KeySearch(**response.json())

    async def get_key(self, key: str) -> Key:
        """Gets information about a specific API key.

        Args:
            key: The key for which to retrieve the information.

        Returns:
            The API key, or `None` if the key is not found.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     keys = await client.get_key("abc123")
        """
        response = await self._http_requests.get(f"keys/{key}")

        return Key(**response.json())

    async def update_key(self, key: KeyUpdate) -> Key:
        """Update an API key.

        Args:
            key: The information to use in updating the key. Note that if an expires_at value
                is included it should be in UTC time.

        Returns:
            The updated API key.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> from meilissearch_async_client.models.client import KeyUpdate
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     key_info = KeyUpdate(
                        key="abc123",
            >>>         indexes=["*"],
            >>>     )
            >>>     keys = await client.update_key(key_info)
        """
        payload = _build_update_key_payload(key, self.json_handler)
        response = await self._http_requests.patch(f"keys/{key.key}", payload)

        return Key(**response.json())

    async def multi_search(
        self,
        queries: list[SearchParams],
        *,
        federation: Federation | FederationMerged | None = None,
        hits_type: Any = JsonDict,
    ) -> list[SearchResultsWithUID] | SearchResultsFederated:
        """Multi-index search.

        Args:
            queries: List of SearchParameters
            federation: If included a single search result with hits built from all queries will
                be returned. This parameter can only be used with Meilisearch >= v1.10.0. Defaults
                to None.
            hits_type: Allows for a custom type to be passed to use for hits. Defaults to
                JsonDict

        Returns:
            Results of the search

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> from meilisearch_python_sdk.models.search import SearchParams
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     queries = [
            >>>         SearchParams(index_uid="my_first_index", query"Some search"),
            >>>         SearchParams(index_uid="my_second_index", query="Another search")
            >>>     ]
            >>>     search_results = await client.search(queries)
        """
        url = "multi-search"
        if federation:
            processed_queries = []
            for query in queries:
                q = query.model_dump(by_alias=True)
                del q["limit"]
                del q["offset"]
                processed_queries.append(q)
        else:
            processed_queries = [x.model_dump(by_alias=True) for x in queries]

        if federation:
            federation_payload = federation.model_dump(by_alias=True)
            if federation.facets_by_index is None:
                del federation_payload["facetsByIndex"]

        else:
            federation_payload = None

        response = await self._http_requests.post(
            url,
            body={
                "federation": federation_payload,
                "queries": processed_queries,
            },
        )

        if federation:
            results = response.json()
            return SearchResultsFederated[hits_type](**results)

        return [SearchResultsWithUID[hits_type](**x) for x in response.json()["results"]]

    async def get_raw_index(self, uid: str) -> IndexInfo | None:
        """Gets the index and returns all the index information rather than an AsyncIndex instance.

        Args:
            uid: The index's unique identifier.

        Returns:
            Index information rather than an AsyncIndex instance.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     index = await client.get_raw_index("movies")
        """
        response = await self.http_client.get(f"indexes/{uid}")

        if response.status_code == 404:
            return None

        return IndexInfo(**response.json())

    async def get_raw_indexes(
        self, *, offset: int | None = None, limit: int | None = None
    ) -> list[IndexInfo] | None:
        """Gets all the indexes.

        Args:
            offset: Number of indexes to skip. The default of None will use the Meilisearch
                default.
            limit: Number of indexes to return. The default of None will use the Meilisearch
                default.

        Returns all the index information rather than an AsyncIndex instance.

        Returns:
            A list of the Index information rather than an AsyncIndex instances.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     index = await client.get_raw_indexes()
        """
        url = _build_offset_limit_url("indexes", offset, limit)
        response = await self._http_requests.get(url)

        if not response.json()["results"]:
            return None

        return [IndexInfo(**x) for x in response.json()["results"]]

    async def get_version(self) -> Version:
        """Get the Meilisearch version.

        Returns:
            Information about the version of Meilisearch.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     version = await client.get_version()
        """
        response = await self._http_requests.get("version")

        return Version(**response.json())

    async def health(self) -> Health:
        """Get health of the Meilisearch server.

        Returns:
            The status of the Meilisearch server.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     health = await client.get_health()
        """
        response = await self._http_requests.get("health")

        return Health(**response.json())

    async def swap_indexes(self, indexes: list[tuple[str, str]]) -> TaskInfo:
        """Swap two indexes.

        Args:
            indexes: A list of tuples, each tuple should contain the indexes to swap.

        Returns:
            The details of the task.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     index = await client.swap_indexes([("index_a", "index_b")])
        """
        processed_indexes = [{"indexes": x} for x in indexes]
        response = await self._http_requests.post("swap-indexes", processed_indexes)

        return TaskInfo(**response.json())

    async def cancel_tasks(
        self,
        *,
        uids: list[str] | None = None,
        index_uids: list[str] | None = None,
        statuses: list[str] | None = None,
        types: list[str] | None = None,
        before_enqueued_at: datetime | None = None,
        after_enqueued_at: datetime | None = None,
        before_started_at: datetime | None = None,
        after_finished_at: datetime | None = None,
    ) -> TaskInfo:
        """Cancel a list of enqueued or processing tasks.

        Defaults to cancelling all tasks.

        Args:
            uids: A list of task UIDs to cancel.
            index_uids: A list of index UIDs for which to cancel tasks.
            statuses: A list of statuses to cancel.
            types: A list of types to cancel.
            before_enqueued_at: Cancel tasks that were enqueued before the specified date time.
            after_enqueued_at: Cancel tasks that were enqueued after the specified date time.
            before_started_at: Cancel tasks that were started before the specified date time.
            after_finished_at: Cancel tasks that were finished after the specified date time.

        Returns:
            The details of the task

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.
            MeilisearchTimeoutError: If the connection times out.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>>
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     await client.cancel_tasks(uids=[1, 2])
        """
        return await _task.async_cancel_tasks(
            self.http_client,
            uids=uids,
            index_uids=index_uids,
            statuses=statuses,
            types=types,
            before_enqueued_at=before_enqueued_at,
            after_enqueued_at=after_enqueued_at,
            before_started_at=before_started_at,
            after_finished_at=after_finished_at,
        )

    async def get_task(self, task_id: int) -> TaskResult:
        """Get a single task from it's task id.

        Args:
            task_id: Identifier of the task to retrieve.

        Returns:
            Results of a task.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.
            MeilisearchTimeoutError: If the connection times out.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> from meilisearch_python_sdk.task import get_task
            >>>
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     await client.get_task(client, 1244)
        """
        return await _task.async_get_task(self.http_client, task_id=task_id)

    async def delete_tasks(
        self,
        *,
        uids: list[str] | None = None,
        index_uids: list[str] | None = None,
        statuses: list[str] | None = None,
        types: list[str] | None = None,
        before_enqueued_at: datetime | None = None,
        after_enqueued_at: datetime | None = None,
        before_started_at: datetime | None = None,
        after_finished_at: datetime | None = None,
    ) -> TaskInfo:
        """Delete a list of tasks.

        Defaults to deleting all tasks.

        Args:
            uids: A list of task UIDs to delete.
            index_uids: A list of index UIDs for which to delete tasks.
            statuses: A list of statuses to delete.
            types: A list of types to delete.
            before_enqueued_at: Delete tasks that were enqueued before the specified date time.
            after_enqueued_at: Delete tasks that were enqueued after the specified date time.
            before_started_at: Delete tasks that were started before the specified date time.
            after_finished_at: Delete tasks that were finished after the specified date time.

        Returns:
            The details of the task

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.
            MeilisearchTimeoutError: If the connection times out.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> from meilisearch_python_sdk.task import delete_tasks
            >>>
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     await client.delete_tasks(uids=[1, 2])
        """
        return await _task.async_delete_tasks(
            self.http_client,
            uids=uids,
            index_uids=index_uids,
            statuses=statuses,
            types=types,
            before_enqueued_at=before_enqueued_at,
            after_enqueued_at=after_enqueued_at,
            before_started_at=before_started_at,
            after_finished_at=after_finished_at,
        )

    async def get_tasks(
        self,
        *,
        index_ids: list[str] | None = None,
        types: str | list[str] | None = None,
    ) -> TaskStatus:
        """Get multiple tasks.

        Args:
            index_ids: A list of index UIDs for which to get the tasks. If provided this will get the
                tasks only for the specified indexes, if not all tasks will be returned. Default = None
            types: Specify specific task types to retrieve. Default = None

        Returns:
            Task statuses.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.
            MeilisearchTimeoutError: If the connection times out.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>>
            >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
            >>>     await client.get_tasks()
        """
        return await _task.async_get_tasks(self.http_client, index_ids=index_ids, types=types)

    async def wait_for_task(
        self,
        task_id: int,
        *,
        timeout_in_ms: int | None = 5000,
        interval_in_ms: int = 50,
        raise_for_status: bool = False,
    ) -> TaskResult:
        """Wait until Meilisearch processes a task, and get its status.

        Args:
            task_id: Identifier of the task to retrieve.
            timeout_in_ms: Amount of time in milliseconds to wait before raising a
                MeilisearchTimeoutError. `None` can also be passed to wait indefinitely. Be aware that
                if the `None` option is used the wait time could be very long. Defaults to 5000.
            interval_in_ms: Time interval in miliseconds to sleep between requests. Defaults to 50.
            raise_for_status: When set to `True` a MeilisearchTaskFailedError will be raised if a task
                has a failed status. Defaults to False.

        Returns:
            Details of the processed update status.

        Raises:
            MeilisearchCommunicationError: If there was an error communicating with the server.
            MeilisearchApiError: If the Meilisearch API returned an error.
            MeilisearchTimeoutError: If the connection times out.
            MeilisearchTaskFailedError: If `raise_for_status` is `True` and a task has a failed status.

        Examples
            >>> from meilisearch_python_sdk import AsyncClient
            >>> >>> documents = [
            >>>     {"id": 1, "title": "Movie 1", "genre": "comedy"},
            >>>     {"id": 2, "title": "Movie 2", "genre": "drama"},
            >>> ]
            >>> async with Client("http://localhost.com", "masterKey") as client:
            >>>     index = client.index("movies")
            >>>     response = await index.add_documents(documents)
            >>>     await client.wait_for_task(client, response.update_id)
        """
        return await _task.async_wait_for_task(
            self.http_client,
            task_id=task_id,
            timeout_in_ms=timeout_in_ms,
            interval_in_ms=interval_in_ms,
            raise_for_status=raise_for_status,
        )

__init__(url, api_key=None, *, timeout=None, verify=True, custom_headers=None, json_handler=None, http2=False)

Class initializer.

Parameters:

Name	Type	Description	Default
url	str	The url to the Meilisearch API (ex: http://localhost:7700)	required
api_key	str | None	The optional API key for Meilisearch. Defaults to None.	None
timeout	int | None	The amount of time in seconds that the client will wait for a response before timing out. Defaults to None.	None
verify	str | bool | SSLContext	SSL certificates (a.k.a CA bundle) used to verify the identity of requested hosts. Either True (default CA bundle), a path to an SSL certificate file, or False (disable verification)	True
custom_headers	dict[str, str] | None	Custom headers to add when sending data to Meilisearch. Defaults to None.	None
json_handler	BuiltinHandler | OrjsonHandler | UjsonHandler | None	The module to use for json operations. The options are BuiltinHandler (uses the json module from the standard library), OrjsonHandler (uses orjson), or UjsonHandler (uses ujson). Note that in order use orjson or ujson the corresponding extra needs to be included. Default: BuiltinHandler.	None
http2	bool	Whether or not to use HTTP/2. Defaults to False.	False

Source code in meilisearch_python_sdk/_client.py

def __init__(
    self,
    url: str,
    api_key: str | None = None,
    *,
    timeout: int | None = None,
    verify: str | bool | SSLContext = True,
    custom_headers: dict[str, str] | None = None,
    json_handler: BuiltinHandler | OrjsonHandler | UjsonHandler | None = None,
    http2: bool = False,
) -> None:
    """Class initializer.

    Args:
        url: The url to the Meilisearch API (ex: http://localhost:7700)
        api_key: The optional API key for Meilisearch. Defaults to None.
        timeout: The amount of time in seconds that the client will wait for a response before
            timing out. Defaults to None.
        verify: SSL certificates (a.k.a CA bundle) used to
            verify the identity of requested hosts. Either `True` (default CA bundle),
            a path to an SSL certificate file, or `False` (disable verification)
        custom_headers: Custom headers to add when sending data to Meilisearch. Defaults to
            None.
        json_handler: The module to use for json operations. The options are BuiltinHandler
            (uses the json module from the standard library), OrjsonHandler (uses orjson), or
            UjsonHandler (uses ujson). Note that in order use orjson or ujson the corresponding
            extra needs to be included. Default: BuiltinHandler.
        http2: Whether or not to use HTTP/2. Defaults to False.
    """
    super().__init__(api_key, custom_headers, json_handler)

    self.http_client = HttpxAsyncClient(
        base_url=url, timeout=timeout, headers=self._headers, verify=verify, http2=http2
    )
    self._http_requests = AsyncHttpRequests(self.http_client, json_handler=self.json_handler)

aclose() async

Closes the client.

This only needs to be used if the client was not created with a context manager.

Source code in meilisearch_python_sdk/_client.py

async def aclose(self) -> None:
    """Closes the client.

    This only needs to be used if the client was not created with a context manager.
    """
    await self.http_client.aclose()

cancel_tasks(*, uids=None, index_uids=None, statuses=None, types=None, before_enqueued_at=None, after_enqueued_at=None, before_started_at=None, after_finished_at=None) async

Cancel a list of enqueued or processing tasks.

Defaults to cancelling all tasks.

Parameters:

Name	Type	Description	Default
uids	list[str] | None	A list of task UIDs to cancel.	None
index_uids	list[str] | None	A list of index UIDs for which to cancel tasks.	None
statuses	list[str] | None	A list of statuses to cancel.	None
types	list[str] | None	A list of types to cancel.	None
before_enqueued_at	datetime | None	Cancel tasks that were enqueued before the specified date time.	None
after_enqueued_at	datetime | None	Cancel tasks that were enqueued after the specified date time.	None
before_started_at	datetime | None	Cancel tasks that were started before the specified date time.	None
after_finished_at	datetime | None	Cancel tasks that were finished after the specified date time.	None

Returns:

Type	Description
TaskInfo	The details of the task

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.
MeilisearchTimeoutError	If the connection times out.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> await client.cancel_tasks(uids=[1, 2])

Source code in meilisearch_python_sdk/_client.py

async def cancel_tasks(
    self,
    *,
    uids: list[str] | None = None,
    index_uids: list[str] | None = None,
    statuses: list[str] | None = None,
    types: list[str] | None = None,
    before_enqueued_at: datetime | None = None,
    after_enqueued_at: datetime | None = None,
    before_started_at: datetime | None = None,
    after_finished_at: datetime | None = None,
) -> TaskInfo:
    """Cancel a list of enqueued or processing tasks.

    Defaults to cancelling all tasks.

    Args:
        uids: A list of task UIDs to cancel.
        index_uids: A list of index UIDs for which to cancel tasks.
        statuses: A list of statuses to cancel.
        types: A list of types to cancel.
        before_enqueued_at: Cancel tasks that were enqueued before the specified date time.
        after_enqueued_at: Cancel tasks that were enqueued after the specified date time.
        before_started_at: Cancel tasks that were started before the specified date time.
        after_finished_at: Cancel tasks that were finished after the specified date time.

    Returns:
        The details of the task

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.
        MeilisearchTimeoutError: If the connection times out.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>>
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     await client.cancel_tasks(uids=[1, 2])
    """
    return await _task.async_cancel_tasks(
        self.http_client,
        uids=uids,
        index_uids=index_uids,
        statuses=statuses,
        types=types,
        before_enqueued_at=before_enqueued_at,
        after_enqueued_at=after_enqueued_at,
        before_started_at=before_started_at,
        after_finished_at=after_finished_at,
    )

create_dump() async

Trigger the creation of a Meilisearch dump.

Returns:

Type	Description
TaskInfo	The details of the task.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples:

>>> from meilisearch_python_sdk import AsyncClient
>>> async with AsyncClient("http://localhost.com", "masterKey") as client:
>>>     await client.create_dump()

Source code in meilisearch_python_sdk/_client.py

async def create_dump(self) -> TaskInfo:
    """Trigger the creation of a Meilisearch dump.

    Returns:
        The details of the task.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples:
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     await client.create_dump()
    """
    response = await self._http_requests.post("dumps")

    return TaskInfo(**response.json())

create_index(uid, primary_key=None, *, settings=None, wait=True, timeout_in_ms=None, plugins=None, hits_type=JsonDict) async

Creates a new index.

Parameters:

Name	Type	Description	Default
uid	str	The index's unique identifier.	required
primary_key	str | None	The primary key of the documents. Defaults to None.	None
settings	MeilisearchSettings | None	Settings for the index. The settings can also be updated independently of creating the index. The advantage to updating them here is updating the settings after adding documents will cause the documents to be re-indexed. Because of this it will be faster to update them before adding documents. Defaults to None (i.e. default Meilisearch index settings).	None
wait	bool	If set to True and settings are being updated, the index will be returned after the settings update has completed. If False it will not wait for settings to complete. Default: True	True
timeout_in_ms	int | None	Amount of time in milliseconds to wait before raising a MeilisearchTimeoutError. None can also be passed to wait indefinitely. Be aware that if the None option is used the wait time could be very long. Defaults to None.	None
plugins	AsyncIndexPlugins | None	Optional plugins can be provided to extend functionality.	None
hits_type	Any	Allows for a custom type to be passed to use for hits. Defaults to JsonDict	JsonDict

Returns:

Type	Description
AsyncIndex	An instance of AsyncIndex containing the information of the newly created index.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples:

>>> from meilisearch_python_sdk import AsyncClient
>>> async with AsyncClient("http://localhost.com", "masterKey") as client:
>>>     index = await client.create_index("movies")

Source code in meilisearch_python_sdk/_client.py

async def create_index(
    self,
    uid: str,
    primary_key: str | None = None,
    *,
    settings: MeilisearchSettings | None = None,
    wait: bool = True,
    timeout_in_ms: int | None = None,
    plugins: AsyncIndexPlugins | None = None,
    hits_type: Any = JsonDict,
) -> AsyncIndex:
    """Creates a new index.

    Args:
        uid: The index's unique identifier.
        primary_key: The primary key of the documents. Defaults to None.
        settings: Settings for the index. The settings can also be updated independently of
            creating the index. The advantage to updating them here is updating the settings after
            adding documents will cause the documents to be re-indexed. Because of this it will be
            faster to update them before adding documents. Defaults to None (i.e. default
            Meilisearch index settings).
        wait: If set to True and settings are being updated, the index will be returned after
            the settings update has completed. If False it will not wait for settings to complete.
            Default: True
        timeout_in_ms: Amount of time in milliseconds to wait before raising a
            MeilisearchTimeoutError. `None` can also be passed to wait indefinitely. Be aware that
            if the `None` option is used the wait time could be very long. Defaults to None.
        plugins: Optional plugins can be provided to extend functionality.
        hits_type: Allows for a custom type to be passed to use for hits. Defaults to
            JsonDict

    Returns:
        An instance of AsyncIndex containing the information of the newly created index.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples:
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     index = await client.create_index("movies")
    """
    return await AsyncIndex.create(
        self.http_client,
        uid,
        primary_key,
        settings=settings,
        wait=wait,
        timeout_in_ms=timeout_in_ms,
        plugins=plugins,
        json_handler=self.json_handler,
        hits_type=hits_type,
    )

create_key(key) async

Creates a new API key.

Parameters:

Name	Type	Description	Default
key	KeyCreate	The information to use in creating the key. Note that if an expires_at value is included it should be in UTC time.	required

Returns:

Type	Description
Key	The new API key.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> from meilissearch_async_client.models.client import KeyCreate >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> key_info = KeyCreate( >>> description="My new key", >>> actions=["search"], >>> indexes=["movies"], >>> ) >>> keys = await client.create_key(key_info)

Source code in meilisearch_python_sdk/_client.py

async def create_key(self, key: KeyCreate) -> Key:
    """Creates a new API key.

    Args:
        key: The information to use in creating the key. Note that if an expires_at value
            is included it should be in UTC time.

    Returns:
        The new API key.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> from meilissearch_async_client.models.client import KeyCreate
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     key_info = KeyCreate(
        >>>         description="My new key",
        >>>         actions=["search"],
        >>>         indexes=["movies"],
        >>>     )
        >>>     keys = await client.create_key(key_info)
    """
    response = await self._http_requests.post(
        "keys", self.json_handler.loads(key.model_dump_json(by_alias=True))
    )  # type: ignore[attr-defined]

    return Key(**response.json())

create_snapshot() async

Trigger the creation of a Meilisearch snapshot.

Returns:

Type	Description
TaskInfo	The details of the task.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples:

>>> from meilisearch_python_sdk import AsyncClient
>>> async with AsyncClient("http://localhost.com", "masterKey") as client:
>>>     await client.create_snapshot()

Source code in meilisearch_python_sdk/_client.py

async def create_snapshot(self) -> TaskInfo:
    """Trigger the creation of a Meilisearch snapshot.

    Returns:
        The details of the task.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples:
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     await client.create_snapshot()
    """
    response = await self._http_requests.post("snapshots")

    return TaskInfo(**response.json())

delete_index_if_exists(uid) async

Deletes an index if it already exists.

Parameters:

Name	Type	Description	Default
uid	str	The index's unique identifier.	required

Returns:

Type	Description
bool	True if an index was deleted for False if not.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples:

>>> from meilisearch_python_sdk import AsyncClient
>>> async with AsyncClient("http://localhost.com", "masterKey") as client:
>>>     await client.delete_index_if_exists()

Source code in meilisearch_python_sdk/_client.py

async def delete_index_if_exists(self, uid: str) -> bool:
    """Deletes an index if it already exists.

    Args:
        uid: The index's unique identifier.

    Returns:
        True if an index was deleted for False if not.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples:
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     await client.delete_index_if_exists()
    """
    response = await self._http_requests.delete(f"indexes/{uid}")
    status = await self.wait_for_task(response.json()["taskUid"], timeout_in_ms=100000)
    if status.status == "succeeded":
        return True
    return False

delete_key(key) async

Deletes an API key.

Parameters:

Name	Type	Description	Default
key	str	The key or uid to delete.	required

Returns:

Type	Description
int	The Response status code. 204 signifies a successful delete.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> await client.delete_key("abc123")

Source code in meilisearch_python_sdk/_client.py

async def delete_key(self, key: str) -> int:
    """Deletes an API key.

    Args:
        key: The key or uid to delete.

    Returns:
        The Response status code. 204 signifies a successful delete.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     await client.delete_key("abc123")
    """
    response = await self._http_requests.delete(f"keys/{key}")
    return response.status_code

delete_tasks(*, uids=None, index_uids=None, statuses=None, types=None, before_enqueued_at=None, after_enqueued_at=None, before_started_at=None, after_finished_at=None) async

Delete a list of tasks.

Defaults to deleting all tasks.

Parameters:

Name	Type	Description	Default
uids	list[str] | None	A list of task UIDs to delete.	None
index_uids	list[str] | None	A list of index UIDs for which to delete tasks.	None
statuses	list[str] | None	A list of statuses to delete.	None
types	list[str] | None	A list of types to delete.	None
before_enqueued_at	datetime | None	Delete tasks that were enqueued before the specified date time.	None
after_enqueued_at	datetime | None	Delete tasks that were enqueued after the specified date time.	None
before_started_at	datetime | None	Delete tasks that were started before the specified date time.	None
after_finished_at	datetime | None	Delete tasks that were finished after the specified date time.	None

Returns:

Type	Description
TaskInfo	The details of the task

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.
MeilisearchTimeoutError	If the connection times out.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> from meilisearch_python_sdk.task import delete_tasks >>> >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> await client.delete_tasks(uids=[1, 2])

Source code in meilisearch_python_sdk/_client.py

async def delete_tasks(
    self,
    *,
    uids: list[str] | None = None,
    index_uids: list[str] | None = None,
    statuses: list[str] | None = None,
    types: list[str] | None = None,
    before_enqueued_at: datetime | None = None,
    after_enqueued_at: datetime | None = None,
    before_started_at: datetime | None = None,
    after_finished_at: datetime | None = None,
) -> TaskInfo:
    """Delete a list of tasks.

    Defaults to deleting all tasks.

    Args:
        uids: A list of task UIDs to delete.
        index_uids: A list of index UIDs for which to delete tasks.
        statuses: A list of statuses to delete.
        types: A list of types to delete.
        before_enqueued_at: Delete tasks that were enqueued before the specified date time.
        after_enqueued_at: Delete tasks that were enqueued after the specified date time.
        before_started_at: Delete tasks that were started before the specified date time.
        after_finished_at: Delete tasks that were finished after the specified date time.

    Returns:
        The details of the task

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.
        MeilisearchTimeoutError: If the connection times out.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> from meilisearch_python_sdk.task import delete_tasks
        >>>
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     await client.delete_tasks(uids=[1, 2])
    """
    return await _task.async_delete_tasks(
        self.http_client,
        uids=uids,
        index_uids=index_uids,
        statuses=statuses,
        types=types,
        before_enqueued_at=before_enqueued_at,
        after_enqueued_at=after_enqueued_at,
        before_started_at=before_started_at,
        after_finished_at=after_finished_at,
    )

get_all_stats() async

Get stats for all indexes.

Returns:

Name	Type	Description
	ClientStats	Information about database size and all indexes.
https	ClientStats	//docs.meilisearch.com/reference/api/stats.html

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> stats = await client.get_all_stats()

Source code in meilisearch_python_sdk/_client.py

async def get_all_stats(self) -> ClientStats:
    """Get stats for all indexes.

    Returns:
        Information about database size and all indexes.
        https://docs.meilisearch.com/reference/api/stats.html

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     stats = await client.get_all_stats()
    """
    response = await self._http_requests.get("stats")

    return ClientStats(**response.json())

get_index(uid) async

Gets a single index based on the uid of the index.

Parameters:

Name	Type	Description	Default
uid	str	The index's unique identifier.	required

Returns:

Type	Description
AsyncIndex	An AsyncIndex instance containing the information of the fetched index.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples:

>>> from meilisearch_python_sdk import AsyncClient
>>> async with AsyncClient("http://localhost.com", "masterKey") as client:
>>>     index = await client.get_index()

Source code in meilisearch_python_sdk/_client.py

async def get_index(self, uid: str) -> AsyncIndex:
    """Gets a single index based on the uid of the index.

    Args:
        uid: The index's unique identifier.

    Returns:
        An AsyncIndex instance containing the information of the fetched index.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples:
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     index = await client.get_index()
    """
    return await AsyncIndex(self.http_client, uid, json_handler=self.json_handler).fetch_info()

get_indexes(*, offset=None, limit=None) async

Get all indexes.

Parameters:

Name	Type	Description	Default
offset	int | None	Number of indexes to skip. The default of None will use the Meilisearch default.	None
limit	int | None	Number of indexes to return. The default of None will use the Meilisearch default.	None

Returns:

Type	Description
list[AsyncIndex] | None	A list of all indexes.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples:

>>> from meilisearch_python_sdk import AsyncClient
>>> async with AsyncClient("http://localhost.com", "masterKey") as client:
>>>     indexes = await client.get_indexes()

Source code in meilisearch_python_sdk/_client.py

async def get_indexes(
    self, *, offset: int | None = None, limit: int | None = None
) -> list[AsyncIndex] | None:
    """Get all indexes.

    Args:
        offset: Number of indexes to skip. The default of None will use the Meilisearch
            default.
        limit: Number of indexes to return. The default of None will use the Meilisearch
            default.

    Returns:
        A list of all indexes.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples:
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     indexes = await client.get_indexes()
    """
    url = _build_offset_limit_url("indexes", offset, limit)
    response = await self._http_requests.get(url)

    if not response.json()["results"]:
        return None

    return [
        AsyncIndex(
            http_client=self.http_client,
            uid=x["uid"],
            primary_key=x["primaryKey"],
            created_at=x["createdAt"],
            updated_at=x["updatedAt"],
            json_handler=self.json_handler,
        )
        for x in response.json()["results"]
    ]

get_key(key) async

Gets information about a specific API key.

Parameters:

Name	Type	Description	Default
key	str	The key for which to retrieve the information.	required

Returns:

Type	Description
Key	The API key, or None if the key is not found.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> keys = await client.get_key("abc123")

Source code in meilisearch_python_sdk/_client.py

async def get_key(self, key: str) -> Key:
    """Gets information about a specific API key.

    Args:
        key: The key for which to retrieve the information.

    Returns:
        The API key, or `None` if the key is not found.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     keys = await client.get_key("abc123")
    """
    response = await self._http_requests.get(f"keys/{key}")

    return Key(**response.json())

get_keys(*, offset=None, limit=None) async

Gets the Meilisearch API keys.

Parameters:

Name	Type	Description	Default
offset	int | None	Number of indexes to skip. The default of None will use the Meilisearch default.	None
limit	int | None	Number of indexes to return. The default of None will use the Meilisearch default.	None

Returns:

Type	Description
KeySearch	API keys.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples from meilisearch_python_sdk import AsyncClient async with AsyncClient("http://localhost.com", "masterKey") as client: keys = await client.get_keys()

Source code in meilisearch_python_sdk/_client.py

async def get_keys(self, *, offset: int | None = None, limit: int | None = None) -> KeySearch:
    """Gets the Meilisearch API keys.

    Args:
        offset: Number of indexes to skip. The default of None will use the Meilisearch
            default.
        limit: Number of indexes to return. The default of None will use the Meilisearch
            default.

    Returns:
        API keys.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        from meilisearch_python_sdk import AsyncClient
        async with AsyncClient("http://localhost.com", "masterKey") as client:
            keys = await client.get_keys()
    """
    url = _build_offset_limit_url("keys", offset, limit)
    response = await self._http_requests.get(url)

    return KeySearch(**response.json())

get_or_create_index(uid, primary_key=None, *, plugins=None, hits_type=JsonDict) async

Get an index, or create it if it doesn't exist.

Parameters:

Name	Type	Description	Default
uid	str	The index's unique identifier.	required
primary_key	str | None	The primary key of the documents. Defaults to None.	None
plugins	AsyncIndexPlugins | None	Optional plugins can be provided to extend functionality.	None
hits_type	Any	Allows for a custom type to be passed to use for hits. Defaults to JsonDict	JsonDict

Returns:

Type	Description
AsyncIndex	An instance of AsyncIndex containing the information of the retrieved or newly created index.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.MeilisearchTimeoutError: If the connection times out.
MeilisearchTimeoutError	If the connection times out.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> index = await client.get_or_create_index("movies")

Source code in meilisearch_python_sdk/_client.py

async def get_or_create_index(
    self,
    uid: str,
    primary_key: str | None = None,
    *,
    plugins: AsyncIndexPlugins | None = None,
    hits_type: Any = JsonDict,
) -> AsyncIndex:
    """Get an index, or create it if it doesn't exist.

    Args:
        uid: The index's unique identifier.
        primary_key: The primary key of the documents. Defaults to None.
        plugins: Optional plugins can be provided to extend functionality.
        hits_type: Allows for a custom type to be passed to use for hits. Defaults to
            JsonDict

    Returns:
        An instance of AsyncIndex containing the information of the retrieved or newly created index.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.MeilisearchTimeoutError: If the connection times out.
        MeilisearchTimeoutError: If the connection times out.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     index = await client.get_or_create_index("movies")
    """
    try:
        index_instance = await self.get_index(uid)
    except MeilisearchApiError as err:
        if "index_not_found" not in err.code:
            raise
        index_instance = await self.create_index(
            uid, primary_key, plugins=plugins, hits_type=hits_type
        )
    return index_instance

get_raw_index(uid) async

Gets the index and returns all the index information rather than an AsyncIndex instance.

Parameters:

Name	Type	Description	Default
uid	str	The index's unique identifier.	required

Returns:

Type	Description
IndexInfo | None	Index information rather than an AsyncIndex instance.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> index = await client.get_raw_index("movies")

Source code in meilisearch_python_sdk/_client.py

async def get_raw_index(self, uid: str) -> IndexInfo | None:
    """Gets the index and returns all the index information rather than an AsyncIndex instance.

    Args:
        uid: The index's unique identifier.

    Returns:
        Index information rather than an AsyncIndex instance.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     index = await client.get_raw_index("movies")
    """
    response = await self.http_client.get(f"indexes/{uid}")

    if response.status_code == 404:
        return None

    return IndexInfo(**response.json())

get_raw_indexes(*, offset=None, limit=None) async

Gets all the indexes.

Parameters:

Name	Type	Description	Default
offset	int | None	Number of indexes to skip. The default of None will use the Meilisearch default.	None
limit	int | None	Number of indexes to return. The default of None will use the Meilisearch default.	None

Returns all the index information rather than an AsyncIndex instance.

Returns:

Type	Description
list[IndexInfo] | None	A list of the Index information rather than an AsyncIndex instances.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> index = await client.get_raw_indexes()

Source code in meilisearch_python_sdk/_client.py

async def get_raw_indexes(
    self, *, offset: int | None = None, limit: int | None = None
) -> list[IndexInfo] | None:
    """Gets all the indexes.

    Args:
        offset: Number of indexes to skip. The default of None will use the Meilisearch
            default.
        limit: Number of indexes to return. The default of None will use the Meilisearch
            default.

    Returns all the index information rather than an AsyncIndex instance.

    Returns:
        A list of the Index information rather than an AsyncIndex instances.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     index = await client.get_raw_indexes()
    """
    url = _build_offset_limit_url("indexes", offset, limit)
    response = await self._http_requests.get(url)

    if not response.json()["results"]:
        return None

    return [IndexInfo(**x) for x in response.json()["results"]]

get_task(task_id) async

Get a single task from it's task id.

Parameters:

Name	Type	Description	Default
task_id	int	Identifier of the task to retrieve.	required

Returns:

Type	Description
TaskResult	Results of a task.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.
MeilisearchTimeoutError	If the connection times out.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> from meilisearch_python_sdk.task import get_task >>> >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> await client.get_task(client, 1244)

Source code in meilisearch_python_sdk/_client.py

async def get_task(self, task_id: int) -> TaskResult:
    """Get a single task from it's task id.

    Args:
        task_id: Identifier of the task to retrieve.

    Returns:
        Results of a task.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.
        MeilisearchTimeoutError: If the connection times out.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> from meilisearch_python_sdk.task import get_task
        >>>
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     await client.get_task(client, 1244)
    """
    return await _task.async_get_task(self.http_client, task_id=task_id)

get_tasks(*, index_ids=None, types=None) async

Get multiple tasks.

Parameters:

Name	Type	Description	Default
index_ids	list[str] | None	A list of index UIDs for which to get the tasks. If provided this will get the tasks only for the specified indexes, if not all tasks will be returned. Default = None	None
types	str | list[str] | None	Specify specific task types to retrieve. Default = None	None

Returns:

Type	Description
TaskStatus	Task statuses.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.
MeilisearchTimeoutError	If the connection times out.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> await client.get_tasks()

Source code in meilisearch_python_sdk/_client.py

async def get_tasks(
    self,
    *,
    index_ids: list[str] | None = None,
    types: str | list[str] | None = None,
) -> TaskStatus:
    """Get multiple tasks.

    Args:
        index_ids: A list of index UIDs for which to get the tasks. If provided this will get the
            tasks only for the specified indexes, if not all tasks will be returned. Default = None
        types: Specify specific task types to retrieve. Default = None

    Returns:
        Task statuses.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.
        MeilisearchTimeoutError: If the connection times out.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>>
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     await client.get_tasks()
    """
    return await _task.async_get_tasks(self.http_client, index_ids=index_ids, types=types)

get_version() async

Get the Meilisearch version.

Returns:

Type	Description
Version	Information about the version of Meilisearch.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> version = await client.get_version()

Source code in meilisearch_python_sdk/_client.py

async def get_version(self) -> Version:
    """Get the Meilisearch version.

    Returns:
        Information about the version of Meilisearch.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     version = await client.get_version()
    """
    response = await self._http_requests.get("version")

    return Version(**response.json())

health() async

Get health of the Meilisearch server.

Returns:

Type	Description
Health	The status of the Meilisearch server.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> health = await client.get_health()

Source code in meilisearch_python_sdk/_client.py

async def health(self) -> Health:
    """Get health of the Meilisearch server.

    Returns:
        The status of the Meilisearch server.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     health = await client.get_health()
    """
    response = await self._http_requests.get("health")

    return Health(**response.json())

index(uid, *, plugins=None)

Create a local reference to an index identified by UID, without making an HTTP call.

Because no network call is made this method is not awaitable.

Parameters:

Name	Type	Description	Default
uid	str	The index's unique identifier.	required
plugins	AsyncIndexPlugins | None	Optional plugins can be provided to extend functionality.	None

Returns:

Type	Description
AsyncIndex	An AsyncIndex instance.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples:

>>> from meilisearch_python_sdk import AsyncClient
>>> async with AsyncClient("http://localhost.com", "masterKey") as client:
>>>     index = client.index("movies")

Source code in meilisearch_python_sdk/_client.py

def index(self, uid: str, *, plugins: AsyncIndexPlugins | None = None) -> AsyncIndex:
    """Create a local reference to an index identified by UID, without making an HTTP call.

    Because no network call is made this method is not awaitable.

    Args:
        uid: The index's unique identifier.
        plugins: Optional plugins can be provided to extend functionality.

    Returns:
        An AsyncIndex instance.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples:
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     index = client.index("movies")
    """
    return AsyncIndex(
        self.http_client, uid=uid, plugins=plugins, json_handler=self.json_handler
    )

multi_search(queries, *, federation=None, hits_type=JsonDict) async

Multi-index search.

Parameters:

Name	Type	Description	Default
queries	list[SearchParams]	List of SearchParameters	required
federation	Federation | FederationMerged | None	If included a single search result with hits built from all queries will be returned. This parameter can only be used with Meilisearch >= v1.10.0. Defaults to None.	None
hits_type	Any	Allows for a custom type to be passed to use for hits. Defaults to JsonDict	JsonDict

Returns:

Type	Description
list[SearchResultsWithUID] | SearchResultsFederated	Results of the search

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> from meilisearch_python_sdk.models.search import SearchParams >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> queries = [ >>> SearchParams(index_uid="my_first_index", query"Some search"), >>> SearchParams(index_uid="my_second_index", query="Another search") >>> ] >>> search_results = await client.search(queries)

Source code in meilisearch_python_sdk/_client.py

async def multi_search(
    self,
    queries: list[SearchParams],
    *,
    federation: Federation | FederationMerged | None = None,
    hits_type: Any = JsonDict,
) -> list[SearchResultsWithUID] | SearchResultsFederated:
    """Multi-index search.

    Args:
        queries: List of SearchParameters
        federation: If included a single search result with hits built from all queries will
            be returned. This parameter can only be used with Meilisearch >= v1.10.0. Defaults
            to None.
        hits_type: Allows for a custom type to be passed to use for hits. Defaults to
            JsonDict

    Returns:
        Results of the search

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> from meilisearch_python_sdk.models.search import SearchParams
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     queries = [
        >>>         SearchParams(index_uid="my_first_index", query"Some search"),
        >>>         SearchParams(index_uid="my_second_index", query="Another search")
        >>>     ]
        >>>     search_results = await client.search(queries)
    """
    url = "multi-search"
    if federation:
        processed_queries = []
        for query in queries:
            q = query.model_dump(by_alias=True)
            del q["limit"]
            del q["offset"]
            processed_queries.append(q)
    else:
        processed_queries = [x.model_dump(by_alias=True) for x in queries]

    if federation:
        federation_payload = federation.model_dump(by_alias=True)
        if federation.facets_by_index is None:
            del federation_payload["facetsByIndex"]

    else:
        federation_payload = None

    response = await self._http_requests.post(
        url,
        body={
            "federation": federation_payload,
            "queries": processed_queries,
        },
    )

    if federation:
        results = response.json()
        return SearchResultsFederated[hits_type](**results)

    return [SearchResultsWithUID[hits_type](**x) for x in response.json()["results"]]

swap_indexes(indexes) async

Swap two indexes.

Parameters:

Name	Type	Description	Default
indexes	list[tuple[str, str]]	A list of tuples, each tuple should contain the indexes to swap.	required

Returns:

Type	Description
TaskInfo	The details of the task.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> index = await client.swap_indexes([("index_a", "index_b")])

Source code in meilisearch_python_sdk/_client.py

async def swap_indexes(self, indexes: list[tuple[str, str]]) -> TaskInfo:
    """Swap two indexes.

    Args:
        indexes: A list of tuples, each tuple should contain the indexes to swap.

    Returns:
        The details of the task.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     index = await client.swap_indexes([("index_a", "index_b")])
    """
    processed_indexes = [{"indexes": x} for x in indexes]
    response = await self._http_requests.post("swap-indexes", processed_indexes)

    return TaskInfo(**response.json())

update_key(key) async

Update an API key.

Parameters:

Name	Type	Description	Default
key	KeyUpdate	The information to use in updating the key. Note that if an expires_at value is included it should be in UTC time.	required

Returns:

Type	Description
Key	The updated API key.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> from meilissearch_async_client.models.client import KeyUpdate >>> async with AsyncClient("http://localhost.com", "masterKey") as client: >>> key_info = KeyUpdate( key="abc123", >>> indexes=["*"], >>> ) >>> keys = await client.update_key(key_info)

Source code in meilisearch_python_sdk/_client.py

async def update_key(self, key: KeyUpdate) -> Key:
    """Update an API key.

    Args:
        key: The information to use in updating the key. Note that if an expires_at value
            is included it should be in UTC time.

    Returns:
        The updated API key.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> from meilissearch_async_client.models.client import KeyUpdate
        >>> async with AsyncClient("http://localhost.com", "masterKey") as client:
        >>>     key_info = KeyUpdate(
                    key="abc123",
        >>>         indexes=["*"],
        >>>     )
        >>>     keys = await client.update_key(key_info)
    """
    payload = _build_update_key_payload(key, self.json_handler)
    response = await self._http_requests.patch(f"keys/{key.key}", payload)

    return Key(**response.json())

wait_for_task(task_id, *, timeout_in_ms=5000, interval_in_ms=50, raise_for_status=False) async

Wait until Meilisearch processes a task, and get its status.

Parameters:

Name	Type	Description	Default
task_id	int	Identifier of the task to retrieve.	required
timeout_in_ms	int | None	Amount of time in milliseconds to wait before raising a MeilisearchTimeoutError. None can also be passed to wait indefinitely. Be aware that if the None option is used the wait time could be very long. Defaults to 5000.	5000
interval_in_ms	int	Time interval in miliseconds to sleep between requests. Defaults to 50.	50
raise_for_status	bool	When set to True a MeilisearchTaskFailedError will be raised if a task has a failed status. Defaults to False.	False

Returns:

Type	Description
TaskResult	Details of the processed update status.

Raises:

Type	Description
MeilisearchCommunicationError	If there was an error communicating with the server.
MeilisearchApiError	If the Meilisearch API returned an error.
MeilisearchTimeoutError	If the connection times out.
MeilisearchTaskFailedError	If raise_for_status is True and a task has a failed status.

Examples >>> from meilisearch_python_sdk import AsyncClient >>> >>> documents = [ >>> {"id": 1, "title": "Movie 1", "genre": "comedy"}, >>> {"id": 2, "title": "Movie 2", "genre": "drama"}, >>> ] >>> async with Client("http://localhost.com", "masterKey") as client: >>> index = client.index("movies") >>> response = await index.add_documents(documents) >>> await client.wait_for_task(client, response.update_id)

Source code in meilisearch_python_sdk/_client.py

async def wait_for_task(
    self,
    task_id: int,
    *,
    timeout_in_ms: int | None = 5000,
    interval_in_ms: int = 50,
    raise_for_status: bool = False,
) -> TaskResult:
    """Wait until Meilisearch processes a task, and get its status.

    Args:
        task_id: Identifier of the task to retrieve.
        timeout_in_ms: Amount of time in milliseconds to wait before raising a
            MeilisearchTimeoutError. `None` can also be passed to wait indefinitely. Be aware that
            if the `None` option is used the wait time could be very long. Defaults to 5000.
        interval_in_ms: Time interval in miliseconds to sleep between requests. Defaults to 50.
        raise_for_status: When set to `True` a MeilisearchTaskFailedError will be raised if a task
            has a failed status. Defaults to False.

    Returns:
        Details of the processed update status.

    Raises:
        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.
        MeilisearchTimeoutError: If the connection times out.
        MeilisearchTaskFailedError: If `raise_for_status` is `True` and a task has a failed status.

    Examples
        >>> from meilisearch_python_sdk import AsyncClient
        >>> >>> documents = [
        >>>     {"id": 1, "title": "Movie 1", "genre": "comedy"},
        >>>     {"id": 2, "title": "Movie 2", "genre": "drama"},
        >>> ]
        >>> async with Client("http://localhost.com", "masterKey") as client:
        >>>     index = client.index("movies")
        >>>     response = await index.add_documents(documents)
        >>>     await client.wait_for_task(client, response.update_id)
    """
    return await _task.async_wait_for_task(
        self.http_client,
        task_id=task_id,
        timeout_in_ms=timeout_in_ms,
        interval_in_ms=interval_in_ms,
        raise_for_status=raise_for_status,
    )