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")
    ...

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,
    ) -> 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)
        """
        super().__init__(api_key)

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

    async def __aenter__(self) -> AsyncClient:
        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,
    ) -> 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.

        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,
        )

    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"],
            )
            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).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)

    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
    ) -> 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.

        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)
        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)
        """
        if is_pydantic_2():
            response = await self._http_requests.post(
                "keys", json.loads(key.model_dump_json(by_alias=True))
            )  # type: ignore[attr-defined]
        else:  # pragma: no cover
            warn(
                "The use of Pydantic less than version 2 is depreciated and will be removed in a future release",
                DeprecationWarning,
                stacklevel=2,
            )
            response = await self._http_requests.post("keys", json.loads(key.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)
        response = await self._http_requests.patch(f"keys/{key.key}", payload)

        return Key(**response.json())

    async def multi_search(self, queries: list[SearchParams]) -> list[SearchResultsWithUID]:
        """Multi-index search.

        Args:

            queries: List of SearchParameters

        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 is_pydantic_2():
            response = await self._http_requests.post(
                url,
                body={"queries": [x.model_dump(by_alias=True) for x in queries]},  # type: ignore[attr-defined]
            )
        else:  # pragma: no cover
            warn(
                "The use of Pydantic less than version 2 is depreciated and will be removed in a future release",
                DeprecationWarning,
                stacklevel=2,
            )
            response = await self._http_requests.post(
                url,
                body={"queries": [x.dict(by_alias=True) for x in queries]},  # type: ignore[attr-defined]
            )

        return [SearchResultsWithUID(**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)

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)

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,
) -> 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)
    """
    super().__init__(api_key)

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

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.

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])

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:

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()

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) async

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.

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")

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,
) -> 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.

    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,
    )

create_key(key) async

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)

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)
    """
    if is_pydantic_2():
        response = await self._http_requests.post(
            "keys", json.loads(key.model_dump_json(by_alias=True))
        )  # type: ignore[attr-defined]
    else:  # pragma: no cover
        warn(
            "The use of Pydantic less than version 2 is depreciated and will be removed in a future release",
            DeprecationWarning,
            stacklevel=2,
        )
        response = await self._http_requests.post("keys", json.loads(key.json(by_alias=True)))  # type: ignore[attr-defined]

    return Key(**response.json())

create_snapshot() async

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()

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.

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()

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.

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")

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.

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])

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:

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()

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.

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()

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).fetch_info()

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

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()

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"],
        )
        for x in response.json()["results"]
    ]

get_key(key) async

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")

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. 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()

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) async

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.

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")

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
) -> 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.

    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)
    return index_instance

get_raw_index(uid) async

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")

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. 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()

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.

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)

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.

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()

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:

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()

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:

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()

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.

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")

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)

multi_search(queries) async

Multi-index search.

Args:

queries: List of SearchParameters

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)

Source code in meilisearch_python_sdk/_client.py

async def multi_search(self, queries: list[SearchParams]) -> list[SearchResultsWithUID]:
    """Multi-index search.

    Args:

        queries: List of SearchParameters

    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 is_pydantic_2():
        response = await self._http_requests.post(
            url,
            body={"queries": [x.model_dump(by_alias=True) for x in queries]},  # type: ignore[attr-defined]
        )
    else:  # pragma: no cover
        warn(
            "The use of Pydantic less than version 2 is depreciated and will be removed in a future release",
            DeprecationWarning,
            stacklevel=2,
        )
        response = await self._http_requests.post(
            url,
            body={"queries": [x.dict(by_alias=True) for x in queries]},  # type: ignore[attr-defined]
        )

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

swap_indexes(indexes) async

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")])

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.

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)

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)
    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.

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)

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,
    )