sync script for mariadb -> algolia

.gitignore

@@ -0,0 +1,6 @@
+.idea/
+
+venv/
+
+.env
+

requirements.txt

@@ -0,0 +1,5 @@
+
+python-dotenv
+
+mysql-connector-python
+algoliasearch==3.0.0

sync-algolia.py

@@ -0,0 +1,70 @@
+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()