Skip to content

Decorators

Decorator Usage

Various decorators are provided that can be used to help with the Meilisearch interaction.

ConnectionInfo

Bases: NamedTuple

Infomation on how to connect to Meilisearch.

url: URL for the Meilisearch server. api_key: The API key for the server.

Source code in meilisearch_python_sdk/decorators.py 
11
12
13
14
15
16
17
18
19
class ConnectionInfo(NamedTuple):
    """Infomation on how to connect to Meilisearch.

    url: URL for the Meilisearch server.
    api_key: The API key for the server.
    """

    url: str
    api_key: str

add_documents(*, index_name, connection_info, batch_size=None, primary_key=None, wait_for_task=False)

Decorator that takes the returned documents from a function and adds them to Meilisearch.

It is required that either an client or url is provided.

Args:

index_name: The name of the index to which the documents should be added.
connection_info: Either an Client instance ConnectionInfo with informtaion on how to
    connect to Meilisearch.
batch_size: If provided the documents will be sent in batches of the specified size.
    Otherwise all documents are sent at once. Default = None.
primary_key: The primary key of the documents. This will be ignored if already set.
    Defaults to None.
wait_for_task: If set to `True` the decorator will wait for the document addition to finish
    indexing before returning, otherwise it will return right away. Default = False.

Returns:

The list of documents proviced by the decorated function.

Raises:

MeilisearchCommunicationError: If there was an error communicating with the server.
MeilisearchApiError: If the Meilisearch API returned an error.
ValueError: If neither an async_client nor an url is provided.

Examples:

>>> from meilisearch_python_sdk import Client
>>> from meilisearch_python_sdk.decorators import add_documents, ConnectionInfo
>>>
>>>
>>> # With `Client`
>>> client = Client(url="http://localhost:7700", api_key="masterKey")
>>> @add_documents(index_name="movies", connection_info=client)
>>> def my_function() -> list[dict[str, Any]]:
>>>     return [{"id": 1, "title": "Test 1"}, {"id": 2, "title": "Test 2"}]
>>>
>>> # With `ConnectionInfo`
>>> @add_documents(
        index_name="movies",
        connection_info=ConnectionInfo(url="http://localhost:7700", api_key="masterKey"),
    )
>>> def my_function() -> list[dict[str, Any]]:
>>>     return [{"id": 1, "title": "Test 1"}, {"id": 2, "title": "Test 2"}]
Source code in meilisearch_python_sdk/decorators.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def add_documents(
    *,
    index_name: str,
    connection_info: Client | ConnectionInfo,
    batch_size: int | None = None,
    primary_key: str | None = None,
    wait_for_task: bool = False,
) -> Callable:
    """Decorator that takes the returned documents from a function and adds them to Meilisearch.

    It is required that either an client or url is provided.

    Args:

        index_name: The name of the index to which the documents should be added.
        connection_info: Either an Client instance ConnectionInfo with informtaion on how to
            connect to Meilisearch.
        batch_size: If provided the documents will be sent in batches of the specified size.
            Otherwise all documents are sent at once. Default = None.
        primary_key: The primary key of the documents. This will be ignored if already set.
            Defaults to None.
        wait_for_task: If set to `True` the decorator will wait for the document addition to finish
            indexing before returning, otherwise it will return right away. Default = False.

    Returns:

        The list of documents proviced by the decorated function.

    Raises:

        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.
        ValueError: If neither an async_client nor an url is provided.

    Examples:

        >>> from meilisearch_python_sdk import Client
        >>> from meilisearch_python_sdk.decorators import add_documents, ConnectionInfo
        >>>
        >>>
        >>> # With `Client`
        >>> client = Client(url="http://localhost:7700", api_key="masterKey")
        >>> @add_documents(index_name="movies", connection_info=client)
        >>> def my_function() -> list[dict[str, Any]]:
        >>>     return [{"id": 1, "title": "Test 1"}, {"id": 2, "title": "Test 2"}]
        >>>
        >>> # With `ConnectionInfo`
        >>> @add_documents(
                index_name="movies",
                connection_info=ConnectionInfo(url="http://localhost:7700", api_key="masterKey"),
            )
        >>> def my_function() -> list[dict[str, Any]]:
        >>>     return [{"id": 1, "title": "Test 1"}, {"id": 2, "title": "Test 2"}]
    """

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            result = func(*args, **kwargs)
            if isinstance(connection_info, Client):
                _add_documents(
                    connection_info,
                    index_name,
                    result,
                    batch_size,
                    primary_key,
                    wait_for_task,
                )
                return result

            decorator_client = Client(url=connection_info.url, api_key=connection_info.api_key)
            _add_documents(
                decorator_client,
                index_name,
                result,
                batch_size,
                primary_key,
                wait_for_task,
            )

            return result

        return wrapper

    return decorator

async_add_documents(*, index_name, connection_info, batch_size=None, primary_key=None, wait_for_task=False)

Decorator that takes the returned documents from a function and asyncronously adds them to Meilisearch.

It is required that either an async_client or url is provided.

Args:

index_name: The name of the index to which the documents should be added.
connection_info: Either an AsyncClient instance ConnectionInfo with informtaion on how to
    connect to Meilisearch.
batch_size: If provided the documents will be sent in batches of the specified size.
    Otherwise all documents are sent at once. Default = None.
primary_key: The primary key of the documents. This will be ignored if already set.
    Defaults to None.
wait_for_task: If set to `True` the decorator will wait for the document addition to finish
    indexing before returning, otherwise it will return right away. Default = False.

Returns:

The list of documents proviced by the decorated function.

Raises:

MeilisearchCommunicationError: If there was an error communicating with the server.
MeilisearchApiError: If the Meilisearch API returned an error.
ValueError: If neither an async_client nor an url is provided.

Examples:

>>> from meilisearch_python_sdk import AsyncClient
>>> from meilisearch_python_sdk.decorators import async_add_documents, ConnectionInfo
>>>
>>>
>>> # with `AsyncClient`
>>> client = AsyncClient(url="http://localhost:7700", api_key="masterKey")
>>> @async_add_documents(index_name="movies", connection_info=client)
>>> async def my_function() -> list[dict[str, Any]]:
>>>     return [{"id": 1, "title": "Test 1"}, {"id": 2, "title": "Test 2"}]
>>>
>>> # with `ConnectionInfo`
>>> @async_add_documents(
        index_name="movies",
        connection_info=ConnectionInfo(url="http://localhost:7700", api_key="masterKey"),
    )
>>> async def my_function() -> list[dict[str, Any]]:
>>>     return [{"id": 1, "title": "Test 1"}, {"id": 2, "title": "Test 2"}]
Source code in meilisearch_python_sdk/decorators.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def async_add_documents(
    *,
    index_name: str,
    connection_info: AsyncClient | ConnectionInfo,
    batch_size: int | None = None,
    primary_key: str | None = None,
    wait_for_task: bool = False,
) -> Callable:
    """Decorator that takes the returned documents from a function and asyncronously adds them to Meilisearch.

    It is required that either an async_client or url is provided.

    Args:

        index_name: The name of the index to which the documents should be added.
        connection_info: Either an AsyncClient instance ConnectionInfo with informtaion on how to
            connect to Meilisearch.
        batch_size: If provided the documents will be sent in batches of the specified size.
            Otherwise all documents are sent at once. Default = None.
        primary_key: The primary key of the documents. This will be ignored if already set.
            Defaults to None.
        wait_for_task: If set to `True` the decorator will wait for the document addition to finish
            indexing before returning, otherwise it will return right away. Default = False.

    Returns:

        The list of documents proviced by the decorated function.

    Raises:

        MeilisearchCommunicationError: If there was an error communicating with the server.
        MeilisearchApiError: If the Meilisearch API returned an error.
        ValueError: If neither an async_client nor an url is provided.

    Examples:

        >>> from meilisearch_python_sdk import AsyncClient
        >>> from meilisearch_python_sdk.decorators import async_add_documents, ConnectionInfo
        >>>
        >>>
        >>> # with `AsyncClient`
        >>> client = AsyncClient(url="http://localhost:7700", api_key="masterKey")
        >>> @async_add_documents(index_name="movies", connection_info=client)
        >>> async def my_function() -> list[dict[str, Any]]:
        >>>     return [{"id": 1, "title": "Test 1"}, {"id": 2, "title": "Test 2"}]
        >>>
        >>> # with `ConnectionInfo`
        >>> @async_add_documents(
                index_name="movies",
                connection_info=ConnectionInfo(url="http://localhost:7700", api_key="masterKey"),
            )
        >>> async def my_function() -> list[dict[str, Any]]:
        >>>     return [{"id": 1, "title": "Test 1"}, {"id": 2, "title": "Test 2"}]
    """

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            result = await func(*args, **kwargs)
            if isinstance(connection_info, AsyncClient):
                await _async_add_documents(
                    connection_info,
                    index_name,
                    result,
                    batch_size,
                    primary_key,
                    wait_for_task,
                )
                return result

            async with AsyncClient(connection_info.url, connection_info.api_key) as client:
                await _async_add_documents(
                    client, index_name, result, batch_size, primary_key, wait_for_task
                )

            return result

        return wrapper

    return decorator