meilisync

Before using meilisync, ensure your source database is configured correctly:

• MySQL: Set binlog_format = ROW and ensure binary logging is enabled.
• PostgreSQL: Set wal_level = logical and install the wal2json extension to enable logical replication.
• MongoDB: Enable replica set mode to allow the use of change streams.

Install meilisync via pip using extras to include support for your specific database type:

• MySQL: pip install meilisync[mysql]
• PostgreSQL: pip install meilisync[postgres]
• MongoDB: pip install meilisync[mongo]
• All databases: pip install meilisync[all]
• Redis progress tracking: pip install meilisync[redis]

You can run meilisync using Docker by mounting your config.yml file into the container. Use the following docker-compose.yml structure:

version: "3"
services:
  meilisync:
    image: long2ice/meilisync
    volumes:
      - ./config.yml:/meilisync/config.yml
    restart: always

The meilisync command-line interface provides several commands to manage synchronization. By default, it looks for a config.yml in the current directory.

Commands

• start: Begins the incremental synchronization process.
• refresh -t <index_name>: Refreshes all data for a specific index by swapping indexes. Note: Stop the sync process before running this to avoid data inconsistency.
• check -t <index_name>: Checks if the data count in the database is consistent with the data in Meilisearch.
• version: Displays the current version of meilisync.

Options

• --config, -c <path>: Specify a custom configuration file path (default: config.yml).

meilisync start
meilisync refresh -t test
meilisync check -t test

Use get_progress(type_: ProgressType) to dynamically retrieve a specific Progress class implementation based on a ProgressType enum value. This is useful for selecting how synchronization progress is monitored or reported.

from meilisync.discover import get_progress
from meilisync.enums import ProgressType

# Retrieve the class implementation for a specific progress type
progress_class = get_progress(ProgressType.SOME_PROGRESS_TYPE)

# Instantiate the progress handler
instance = progress_class()

The load_plugin function allows you to dynamically instantiate a plugin class using its full Python import path (e.g., package.module.ClassName). It uses importlib to import the module and then retrieves the specified class.

from meilisync.plugin import load_plugin

# Example usage: loading a class named 'MyPlugin' from 'my_package.plugins'
plugin_class = load_plugin("my_package.plugins.MyPlugin")
plugin_instance = plugin_class()

Use handle_event to process a single Event for a specific Sync configuration. It executes the following sequence:

1. Runs pre_event hooks on all registered plugins.
2. Performs the Meilisearch operation (add_documents, update_documents, or delete_documents) based on the EventType.
3. Runs post_event hooks on all registered plugins.

# event is an instance of Event, sync is an instance of Sync
await meili.handle_event(event, sync)

The Meili class is the primary entry point for interacting with Meilisearch. It manages an asynchronous client and a collection of plugins that can intercept events.

To initialize it, provide the api_url, api_key, and optionally a list of plugins (either instances of Plugin or their classes) and a wait_for_task_timeout in milliseconds.

from meilisync.meili import Meili

meili = Meili(
    api_url="http://localhost:7700",
    api_key="masterKey",
    plugins=[MyPlugin],
    wait_for_task_timeout=30000
)

The refresh_data method performs an atomic index swap to refresh an entire index without downtime.

It follows this workflow:

1. Creates a temporary index named {index_name}_tmp.
2. Copies the settings from the original index to the temporary index.
3. Iterates through the provided AsyncGenerator of data, calling add_data for each batch.
4. Waits for all insertion tasks to complete.
5. Swaps the original index with the temporary index using swap_indexes.
6. Deletes the temporary index.

This is the recommended way to perform full re-indexes.

async def my_data_generator():
    yield [{"id": 1, "name": "item1"}]
    yield [{"id": 2, "name": "item2"}]

# sync is an instance of Sync configuration
count = await meili.refresh_data(sync, my_data_generator())
print(f"Refreshed {count} documents")

The index_exists method checks for the existence of a specific index in Meilisearch. It returns True if the index is found and False if the API returns a index_not_found error. Other errors are re-raised.

exists = await meili.index_exists("my_index")
if exists:
    print("Index is ready")