PythonMicroservices/sync-algolia.py

import os

from algoliasearch.search_client import SearchClient
from dotenv import load_dotenv

import mysql.connector


load_dotenv()

MARIADB_CONFIG = {
    'host': os.getenv('MARIADB_HOSTNAME'),
    'user': os.getenv('MARIADB_USERNAME'),
    'password': os.getenv('MARIADB_PASSWORD'),
    'database': os.getenv('MARIADB_DATABASE')
}
ALGOLIA_APP_ID = os.getenv('ALGOLIA_APP_ID')
ALGOLIA_API_KEY = os.getenv('ALGOLIA_API_ADMIN_KEY')

MARIADB_CONNECTION = mysql.connector.connect(**MARIADB_CONFIG)
MARIADB_CURSOR = MARIADB_CONNECTION.cursor(dictionary=True)

ALGOLIA_CLIENT = SearchClient.create(ALGOLIA_APP_ID, ALGOLIA_API_KEY)

def fetch_mariadb_data(table_name: str):
    query = f"SELECT ID,Title FROM {table_name};"
    MARIADB_CURSOR.execute(query)
    return MARIADB_CURSOR.fetchall()


def sync(index_name: str, table_name: str):
    index = ALGOLIA_CLIENT.init_index(index_name)

    db_data = fetch_mariadb_data(table_name=table_name)
    print(db_data)
    db_dict = {str(row['ID']): row for row in db_data}

    algolia_objects = []
    for hit in index.browse_objects():
        algolia_objects.append(hit)

    algolia_dict = {obj["objectID"]: obj for obj in algolia_objects}

    to_upsert = []
    for db_id, db_row in db_dict.items():
        if db_id not in algolia_dict or db_row != algolia_dict[db_id]:
            db_row["objectID"] = db_id  # Algolia requires "objectID"
            to_upsert.append({
                "objectID": db_id,
                "title": db_row["Title"],
            })

    to_delete = [obj_id for obj_id in algolia_dict if obj_id not in db_dict]

    if to_upsert:
        index.save_objects(to_upsert)
        print(f"Upserted {len(to_upsert)} objects.")

    if to_delete:
        index.delete_objects(to_delete)
        print(f"Deleted {len(to_delete)} objects.")


if __name__ == "__main__":
    try:
        sync('FolkTunes_LIVE', 'Tunes')
        sync('Dances_LIVE', 'Dances')
    finally:
        MARIADB_CURSOR.close()
        MARIADB_CONNECTION.close()