Skip to content

Plugins

Plugins can be used to extend the functionality of certain methods, currently plugins are only supported for indexes. To create plugins you creat a class that implements the Protocol for the plugin, then add an instance of your class to the plugins when creating an index. Passing protocols is done through a named tuple that specifies where the plugin should run. The options are:

  • add_documents_plugins: Runs the plugins when adding documents. This runs for all the add documents methods, i.e. add_documents_in_batches.
  • delete_all_documents_plugins: Run on the delete_all_documents method.
  • delete_document_plugins: Run on the delete_document method.
  • delete_documents_plugins: Run on the delete_documents method.
  • delete_documents_by_filter_plugins: Run on the delete_documents_by_filter method.
  • search_plugins: Run on the search and facet_search methods.
  • update_documents_plugins: Run on the update_document method.

When creating your plugin you specify if you want it to run before or after the default functionality. Additionally plugins for async indexes can be run concurrently with the default functionality.

Examples

Search metrics

It is common to want to know what users are searching for, however Meilisearch doesn't provide a way to track this out of the box. A search plugin could be used to implement this functionality yourself.

Note that in these examples the protocol is satisfied by providing the CONNECURRENT_EVENT, POST_EVENT, and PRE_EVENT variables and the async def run_plugin(self, event: AsyncEvent, **kwargs: Any) -> None: method for an async index, or the POST_EVENT and PRE_EVENT variables , and def run_plugin(self, event: Event, **kwargs: Any) -> None: method for a non-async index. You class can contain any additional methods/variables needed as long as the protocol requirements have been satisfied.

Async index

import asyncio
import json
import sqlite3
from typing import Any

from meilisearch_python_sdk import AsyncClient
from meilisearch_python_sdk.plugins import AsyncEvent, AsyncIndexPlugins


class SearchTrackerPlugin:
    CONCURRENT_EVENT = True  # Specifies the plugin should be run concurrently with the search
    POST_EVENT = False
    PRE_EVENT = False

    def __init__(self) -> None:
        self.conn = sqlite3.Connection("examples/search_tracker.db")
        self.create_table()

    def create_table(self) -> None:
        try:
            cursor = self.conn.cursor()
            cursor.execute("CREATE TABLE IF NOT EXISTS searches(query STRING)")
        finally:
            cursor.close()

    async def run_plugin(self, event: AsyncEvent, **kwargs: Any) -> None:
        """Note that this example uses sqlite which does not provide an async driver.

        Typically if you are using the AsyncClient you would also be using an async driver for the
        database. sqlite is used in this example for simplicity.
        """
        if kwargs.get("query"):
            self.save_search_query(kwargs["query"])

    def save_search_query(self, query: str) -> None:
        try:
            cursor = self.conn.cursor()
            cursor.execute("INSERT INTO searches VALUES(?)", (query,))
            self.conn.commit()
        finally:
            cursor.close()


async def main() -> int:
    with open("datasets/small_movies.json") as f:
        documents = json.load(f)

    async with AsyncClient("http://127.0.0.1:7700", "masterKey") as client:
        plugins = AsyncIndexPlugins(search_plugins=(SearchTrackerPlugin(),))
        index = await client.create_index("movies", primary_key="id", plugins=plugins)
        task = await index.add_documents(documents)
        await client.wait_for_task(task.task_uid)
        result = await index.search("Cars")
        print(result)  # noqa: T201

    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))

Index

import json
import sqlite3
from typing import Any

from meilisearch_python_sdk import Client
from meilisearch_python_sdk.plugins import Event, IndexPlugins


class SearchTrackerPlugin:
    POST_EVENT = False
    PRE_EVENT = True  # Specifies the plugin should be run before the search

    def __init__(self) -> None:
        self.conn = sqlite3.Connection("examples/search_tracker.db")
        self.create_table()

    def create_table(self) -> None:
        try:
            cursor = self.conn.cursor()
            cursor.execute("CREATE TABLE IF NOT EXISTS searches(query STRING)")
        finally:
            cursor.close()

    def run_plugin(self, event: Event, **kwargs: Any) -> None:
        if kwargs.get("query"):
            self.save_search_query(kwargs["query"])

    def save_search_query(self, query: str) -> None:
        try:
            cursor = self.conn.cursor()
            cursor.execute("INSERT INTO searches VALUES(?)", (query,))
            self.conn.commit()
        finally:
            cursor.close()


def main() -> int:
    with open("datasets/small_movies.json") as f:
        documents = json.load(f)

    with Client("http://127.0.0.1:7700", "masterKey") as client:
        plugins = IndexPlugins(search_plugins=(SearchTrackerPlugin(),))
        index = client.create_index("movies", primary_key="id", plugins=plugins)
        task = index.add_documents(documents)
        client.wait_for_task(task.task_uid)
        result = index.search("Cars")
        print(result)  # noqa: T201

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Modify documents and search results

A pre event plugin can be used to modify the documents before sending for indexing. In this example a new access field will be added to the documents before they are added or updated. The example will set every other record to admin access with the other records being set to read. This will illustrate the idea of modifying documents even it if doesn't make real world sense.

A post search plugin, this type of search plugin can only be used post search because it requires the result of the search, will be used to remove records marked as admin before returning the result. In the real world this filtering would probably be done with a filterable field in Meilisearch,but again, this is just used here to illustrate the idea.

Async Index

import asyncio
import json
from typing import Any, Sequence

from meilisearch_python_sdk import AsyncClient
from meilisearch_python_sdk.models.search import SearchResults
from meilisearch_python_sdk.plugins import AsyncEvent, AsyncIndexPlugins
from meilisearch_python_sdk.types import JsonMapping


class ModifyDocumentPlugin:
    CONCURRENT_EVENT = False
    POST_EVENT = False
    PRE_EVENT = True  # Specifies the plugin should be run before adding documents

    async def run_document_plugin(
        self, event: AsyncEvent, *, documents: Sequence[JsonMapping], **kwargs: Any
    ) -> Sequence[JsonMapping]:
        updated = []
        for i, document in enumerate(documents):
            if i % 2 == 0:
                document["access"] = "admin"
            else:
                document["access"] = "read"

            updated.append(document)

        return updated


class FilterSearchResultsPlugin:
    CONCURRENT_EVENT = False
    POST_EVENT = True  # Specifies the plugin should be run after the search
    PRE_EVENT = False

    async def run_post_search_plugin(
        self, event: AsyncEvent, *, search_results: SearchResults, **kwargs: Any
    ) -> SearchResults:
        filtered_hits = []
        for hit in search_results.hits:
            if hit["access"] != "admin":
                filtered_hits.append(hit)

        search_results.hits = filtered_hits

        return search_results


async def main() -> int:
    with open("datasets/small_movies.json") as f:
        documents = json.load(f)

    async with AsyncClient("http://127.0.0.1:7700", "masterKey") as client:
        plugins = AsyncIndexPlugins(
            add_documents_plugins=(ModifyDocumentPlugin(),),
            update_documents_plugins=(ModifyDocumentPlugin(),),
            search_plugins=(FilterSearchResultsPlugin(),),
        )
        index = await client.create_index("movies", primary_key="id", plugins=plugins)
        task = await index.add_documents(documents)
        await client.wait_for_task(task.task_uid)
        result = await index.search("cars")
        print(result)  # noqa: T201

    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))

Index

import json
from typing import Any, Sequence

from meilisearch_python_sdk import Client
from meilisearch_python_sdk.models.search import SearchResults
from meilisearch_python_sdk.plugins import Event, IndexPlugins
from meilisearch_python_sdk.types import JsonMapping


class ModifyDocumentPlugin:
    POST_EVENT = False
    PRE_EVENT = True  # Specifies the plugin should be run before adding documents

    def run_document_plugin(
        self, event: Event, *, documents: Sequence[JsonMapping], **kwargs: Any
    ) -> Sequence[JsonMapping]:
        updated = []
        for i, document in enumerate(documents):
            if i % 2 == 0:
                document["access"] = "admin"
            else:
                document["access"] = "read"

            updated.append(document)

        return updated


class FilterSearchResultsPlugin:
    POST_EVENT = True  # Specifies the plugin should be run after the search
    PRE_EVENT = False

    def run_post_search_plugin(
        self, event: Event, *, search_results: SearchResults, **kwargs: Any
    ) -> SearchResults:
        filtered_hits = []
        for hit in search_results.hits:
            if hit["access"] != "admin":
                filtered_hits.append(hit)

        search_results.hits = filtered_hits

        return search_results


def main() -> int:
    with open("datasets/small_movies.json") as f:
        documents = json.load(f)

    with Client("http://127.0.0.1:7700", "masterKey") as client:
        plugins = IndexPlugins(
            add_documents_plugins=(ModifyDocumentPlugin(),),
            update_documents_plugins=(ModifyDocumentPlugin(),),
            search_plugins=(FilterSearchResultsPlugin(),),
        )
        index = client.create_index("movies", primary_key="id", plugins=plugins)
        task = index.add_documents(documents)
        client.wait_for_task(task.task_uid)
        result = index.search("cars")
        print(result)  # noqa: T201

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Plugins API

meilisearch_python_sdk.plugins

AsyncEvent

Bases: Enum

Source code in meilisearch_python_sdk/plugins.py 
13
14
15
16
class AsyncEvent(Enum):
    PRE = "pre"
    CONCURRENT = "concurrent"
    POST = "post"

PRE class-attribute instance-attribute 

PRE = 'pre'

CONCURRENT class-attribute instance-attribute 

CONCURRENT = 'concurrent'

POST class-attribute instance-attribute 

POST = 'post'

Event

Bases: Enum

Source code in meilisearch_python_sdk/plugins.py 
19
20
21
class Event(Enum):
    PRE = "pre"
    POST = "post"

PRE class-attribute instance-attribute 

PRE = 'pre'

POST class-attribute instance-attribute 

POST = 'post'

AsyncPlugin

Bases: Protocol

Source code in meilisearch_python_sdk/plugins.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
class AsyncPlugin(Protocol):
    CONCURRENT_EVENT: bool
    POST_EVENT: bool
    PRE_EVENT: bool

    async def run_plugin(
        self,
        event: AsyncEvent,
        **kwargs: Any,  # noqa: ANN401
    ) -> (
        None | list[JsonDict] | TaskInfo | list[TaskInfo] | SearchResults | FacetSearchResults
    ):  # pragma: no cover
        ...

CONCURRENT_EVENT instance-attribute 

CONCURRENT_EVENT: bool

POST_EVENT instance-attribute 

POST_EVENT: bool

PRE_EVENT instance-attribute 

PRE_EVENT: bool

run_plugin async 

run_plugin(
    event: AsyncEvent, **kwargs: Any
) -> (
    None
    | list[JsonDict]
    | TaskInfo
    | list[TaskInfo]
    | SearchResults
    | FacetSearchResults
)
Source code in meilisearch_python_sdk/plugins.py 
29
30
31
32
33
34
35
36
async def run_plugin(
    self,
    event: AsyncEvent,
    **kwargs: Any,  # noqa: ANN401
) -> (
    None | list[JsonDict] | TaskInfo | list[TaskInfo] | SearchResults | FacetSearchResults
):  # pragma: no cover
    ...

AsyncDocumentPlugin

Bases: Protocol

Source code in meilisearch_python_sdk/plugins.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class AsyncDocumentPlugin(Protocol):
    CONCURRENT_EVENT: bool
    POST_EVENT: bool
    PRE_EVENT: bool

    async def run_document_plugin(
        self,
        event: AsyncEvent,
        *,
        documents: Sequence[JsonMapping],
        primary_key: str | None,
        **kwargs: Any,  # noqa: ANN401
    ) -> Sequence[JsonMapping] | None:  # pragma: no cover
        ...

CONCURRENT_EVENT instance-attribute 

CONCURRENT_EVENT: bool

POST_EVENT instance-attribute 

POST_EVENT: bool

PRE_EVENT instance-attribute 

PRE_EVENT: bool

run_document_plugin async 

run_document_plugin(
    event: AsyncEvent,
    *,
    documents: Sequence[JsonMapping],
    primary_key: str | None,
    **kwargs: Any,
) -> Sequence[JsonMapping] | None
Source code in meilisearch_python_sdk/plugins.py 
44
45
46
47
48
49
50
51
52
async def run_document_plugin(
    self,
    event: AsyncEvent,
    *,
    documents: Sequence[JsonMapping],
    primary_key: str | None,
    **kwargs: Any,  # noqa: ANN401
) -> Sequence[JsonMapping] | None:  # pragma: no cover
    ...

AsyncPostSearchPlugin

Bases: Protocol

Source code in meilisearch_python_sdk/plugins.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
class AsyncPostSearchPlugin(Protocol):
    CONCURRENT_EVENT: bool
    POST_EVENT: bool
    PRE_EVENT: bool

    async def run_post_search_plugin(
        self,
        event: AsyncEvent,
        *,
        search_results: SearchResults,
        **kwargs: Any,  # noqa: ANN401
    ) -> SearchResults | None:  # pragma: no cover
        ...

CONCURRENT_EVENT instance-attribute 

CONCURRENT_EVENT: bool

POST_EVENT instance-attribute 

POST_EVENT: bool

PRE_EVENT instance-attribute 

PRE_EVENT: bool

run_post_search_plugin async 

run_post_search_plugin(
    event: AsyncEvent,
    *,
    search_results: SearchResults,
    **kwargs: Any,
) -> SearchResults | None
Source code in meilisearch_python_sdk/plugins.py 
60
61
62
63
64
65
66
67
async def run_post_search_plugin(
    self,
    event: AsyncEvent,
    *,
    search_results: SearchResults,
    **kwargs: Any,  # noqa: ANN401
) -> SearchResults | None:  # pragma: no cover
    ...

Plugin

Bases: Protocol

Source code in meilisearch_python_sdk/plugins.py 
70
71
72
73
74
75
76
77
78
79
80
81
class Plugin(Protocol):
    POST_EVENT: bool
    PRE_EVENT: bool

    def run_plugin(
        self,
        event: Event,
        **kwargs: Any,  # noqa: ANN401
    ) -> (
        None | list[JsonDict] | TaskInfo | list[TaskInfo] | SearchResults | FacetSearchResults
    ):  # pragma: no cover
        ...

POST_EVENT instance-attribute 

POST_EVENT: bool

PRE_EVENT instance-attribute 

PRE_EVENT: bool

run_plugin 

run_plugin(
    event: Event, **kwargs: Any
) -> (
    None
    | list[JsonDict]
    | TaskInfo
    | list[TaskInfo]
    | SearchResults
    | FacetSearchResults
)
Source code in meilisearch_python_sdk/plugins.py 
74
75
76
77
78
79
80
81
def run_plugin(
    self,
    event: Event,
    **kwargs: Any,  # noqa: ANN401
) -> (
    None | list[JsonDict] | TaskInfo | list[TaskInfo] | SearchResults | FacetSearchResults
):  # pragma: no cover
    ...

DocumentPlugin

Bases: Protocol

Source code in meilisearch_python_sdk/plugins.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
class DocumentPlugin(Protocol):
    POST_EVENT: bool
    PRE_EVENT: bool

    def run_document_plugin(
        self,
        event: Event,
        *,
        documents: Sequence[JsonMapping],
        primary_key: str | None,
        **kwargs: Any,  # noqa: ANN401
    ) -> Sequence[JsonMapping] | None:  # pragma: no cover
        ...

POST_EVENT instance-attribute 

POST_EVENT: bool

PRE_EVENT instance-attribute 

PRE_EVENT: bool

run_document_plugin 

run_document_plugin(
    event: Event,
    *,
    documents: Sequence[JsonMapping],
    primary_key: str | None,
    **kwargs: Any,
) -> Sequence[JsonMapping] | None
Source code in meilisearch_python_sdk/plugins.py 
88
89
90
91
92
93
94
95
96
def run_document_plugin(
    self,
    event: Event,
    *,
    documents: Sequence[JsonMapping],
    primary_key: str | None,
    **kwargs: Any,  # noqa: ANN401
) -> Sequence[JsonMapping] | None:  # pragma: no cover
    ...

PostSearchPlugin

Bases: Protocol

Source code in meilisearch_python_sdk/plugins.py 
 99
100
101
102
103
104
105
106
107
108
109
110
class PostSearchPlugin(Protocol):
    POST_EVENT: bool
    PRE_EVENT: bool

    def run_post_search_plugin(
        self,
        event: Event,
        *,
        search_results: SearchResults,
        **kwargs: Any,  # noqa: ANN401
    ) -> SearchResults | None:  # pragma: no cover
        ...

POST_EVENT instance-attribute 

POST_EVENT: bool

PRE_EVENT instance-attribute 

PRE_EVENT: bool

run_post_search_plugin 

run_post_search_plugin(
    event: Event,
    *,
    search_results: SearchResults,
    **kwargs: Any,
) -> SearchResults | None
Source code in meilisearch_python_sdk/plugins.py 
103
104
105
106
107
108
109
110
def run_post_search_plugin(
    self,
    event: Event,
    *,
    search_results: SearchResults,
    **kwargs: Any,  # noqa: ANN401
) -> SearchResults | None:  # pragma: no cover
    ...

AsyncIndexPlugins

Bases: NamedTuple

Source code in meilisearch_python_sdk/plugins.py 
113
114
115
116
117
118
119
120
121
class AsyncIndexPlugins(NamedTuple):
    add_documents_plugins: Sequence[AsyncPlugin | AsyncDocumentPlugin] | None = None
    delete_all_documents_plugins: Sequence[AsyncPlugin] | None = None
    delete_document_plugins: Sequence[AsyncPlugin] | None = None
    delete_documents_plugins: Sequence[AsyncPlugin] | None = None
    delete_documents_by_filter_plugins: Sequence[AsyncPlugin] | None = None
    facet_search_plugins: Sequence[AsyncPlugin] | None = None
    search_plugins: Sequence[AsyncPlugin | AsyncPostSearchPlugin] | None = None
    update_documents_plugins: Sequence[AsyncPlugin | AsyncDocumentPlugin] | None = None

add_documents_plugins class-attribute instance-attribute 

add_documents_plugins: (
    Sequence[AsyncPlugin | AsyncDocumentPlugin] | None
) = None

delete_all_documents_plugins class-attribute instance-attribute 

delete_all_documents_plugins: (
    Sequence[AsyncPlugin] | None
) = None

delete_document_plugins class-attribute instance-attribute 

delete_document_plugins: Sequence[AsyncPlugin] | None = None

delete_documents_plugins class-attribute instance-attribute 

delete_documents_plugins: Sequence[AsyncPlugin] | None = (
    None
)

delete_documents_by_filter_plugins class-attribute instance-attribute 

delete_documents_by_filter_plugins: (
    Sequence[AsyncPlugin] | None
) = None

facet_search_plugins class-attribute instance-attribute 

facet_search_plugins: Sequence[AsyncPlugin] | None = None

search_plugins class-attribute instance-attribute 

search_plugins: (
    Sequence[AsyncPlugin | AsyncPostSearchPlugin] | None
) = None

update_documents_plugins class-attribute instance-attribute 

update_documents_plugins: (
    Sequence[AsyncPlugin | AsyncDocumentPlugin] | None
) = None

IndexPlugins

Bases: NamedTuple

Source code in meilisearch_python_sdk/plugins.py 
124
125
126
127
128
129
130
131
132
class IndexPlugins(NamedTuple):
    add_documents_plugins: Sequence[Plugin | DocumentPlugin] | None = None
    delete_all_documents_plugins: Sequence[Plugin] | None = None
    delete_document_plugins: Sequence[Plugin] | None = None
    delete_documents_plugins: Sequence[Plugin] | None = None
    delete_documents_by_filter_plugins: Sequence[Plugin] | None = None
    facet_search_plugins: Sequence[Plugin] | None = None
    search_plugins: Sequence[Plugin | PostSearchPlugin] | None = None
    update_documents_plugins: Sequence[Plugin | DocumentPlugin] | None = None

add_documents_plugins class-attribute instance-attribute 

add_documents_plugins: (
    Sequence[Plugin | DocumentPlugin] | None
) = None

delete_all_documents_plugins class-attribute instance-attribute 

delete_all_documents_plugins: Sequence[Plugin] | None = None

delete_document_plugins class-attribute instance-attribute 

delete_document_plugins: Sequence[Plugin] | None = None

delete_documents_plugins class-attribute instance-attribute 

delete_documents_plugins: Sequence[Plugin] | None = None

delete_documents_by_filter_plugins class-attribute instance-attribute 

delete_documents_by_filter_plugins: (
    Sequence[Plugin] | None
) = None

facet_search_plugins class-attribute instance-attribute 

facet_search_plugins: Sequence[Plugin] | None = None

search_plugins class-attribute instance-attribute 

search_plugins: (
    Sequence[Plugin | PostSearchPlugin] | None
) = None

update_documents_plugins class-attribute instance-attribute 

update_documents_plugins: (
    Sequence[Plugin | DocumentPlugin] | None
) = None