Initialize database and sequence number retrieval

src/floridayvine/__main__.py

@@ -2,6 +2,10 @@
 from typing_extensions import Annotated
 import typer
 from importlib.metadata import version
+from .persistence import (
+    initialize_database,
+    print_sync_status as persistence_print_sync_status,
+)
 from .minio import MinioClient
 from .floriday.misc import (
     get_organizations,
@@ -76,6 +80,21 @@ def sync_organizations(start_seq_number: int = 0, limit_result: int = 5):
     misc.sync_organizations(start_seq_number, limit_result)
 
 
+@app.command()
+def sync_trade_items(start_seq_number: int = 0, limit_result: int = 5):
+    misc.sync_trade_items(start_seq_number, limit_result)
+
+
+@app.command()
+def print_sync_status():
+    persistence_print_sync_status()
+
+
+@app.command()
+def init_db():
+    initialize_database()
+
+
 @app.callback()
 def common(
     ctx: typer.Context,

src/floridayvine/floriday/misc.py

@@ -66,11 +66,13 @@ def sync_organizations(start_seq_number=0, limit_result=5):
         orgs_sync_result = api.get_organizations_by_sequence_number(
             sequence_number=my_sequence, limit_result=limit_result
         )
+        max_seq_nr = orgs_sync_result.maximum_sequence_number
         for org in orgs_sync_result.results:
             print(f"Seq nr {org.sequence_number}: Persisting {org.name} ...")
             persist("organizations", org.organization_id, org.to_dict())
         my_sequence = orgs_sync_result.maximum_sequence_number
         time.sleep(0.5)
+    print("Done syncing organizations")
 
 
 def get_trade_items():
@@ -85,16 +87,24 @@ def get_direct_sales():
     return items
 
 
-def sync_trade_items(base_sync_number=0):
-    access_token = get_access_token()
-    url = f"{BASE_URL}/trade-items/sync/{base_sync_number}"
+def sync_trade_items(start_seq_number=0, limit_result=5):
+    api = TradeItemsApi(_clt)
+    my_sequence = start_seq_number
+    max_seq_nr = api.get_trade_items_max_sequence()
 
-    headers = {
-        "X-Api-Key": API_KEY,
-        "Accept": "application/json",
-        "Authorization": f"Bearer {access_token}",
-    }
+    print(f"Syncing trade items from {my_sequence} to {max_seq_nr} ...")
 
-    response = requests.get(url, headers=headers)
-    response.raise_for_status()
-    return response.json()
+    while my_sequence < max_seq_nr:
+        trade_items_sync_result = api.get_trade_items_by_sequence_number(
+            sequence_number=my_sequence, limit_result=limit_result
+        )
+        max_seq_nr = trade_items_sync_result.maximum_sequence_number
+        for item in trade_items_sync_result.results:
+            print(
+                f"Seq nr {item.sequence_number}: Persisting {item.trade_item_name} ..."
+            )
+            persist("trade_items", item.trade_item_id, item.to_dict())
+        my_sequence = trade_items_sync_result.maximum_sequence_number
+        print(f"Next sequence number: {my_sequence}")
+        time.sleep(0.5)
+    print("Done syncing trade items")

src/floridayvine/persistence.py

@@ -1,17 +1,65 @@
+import os
 from pymongo import MongoClient
 
+mongodb_connection_string = os.getenv("MONGODB_CONNECTION_STRING")
+
+import re
+
+# Regular expression pattern to match and mask the password
+pattern = r"(mongodb://[^:]+:)([^@]+)(@.+)"
+masked_connection_string = re.sub(pattern, r"\1'*****'\3", mongodb_connection_string)
+
+DATABASE = "floriday"
+SYNC_COLLECTIONS = ["organizations", "trade_items"]
+
+
+def initialize_database():
+    """
+    Initialize the database.
+    """
+    print(f"Initializing database on {masked_connection_string}...")
+    with MongoClient(mongodb_connection_string) as client:
+        db = client[DATABASE]
+        existing_collections = set(db.list_collection_names())
+        for collection_name in SYNC_COLLECTIONS:
+            if collection_name not in existing_collections:
+                db.create_collection(collection_name)
+                print(f"Created collection: {collection_name}")
+            collection = db[collection_name]
+            # Create a descending index on the sequence_number field so that we can easily
+            # retrieve the maximum sequence number for a collection to use as the base of
+            # the synchronization process.
+            collection.create_index([("sequence_number", -1)])
+
+
+def get_max_sequence_number(collection_name: str):
+    with MongoClient(mongodb_connection_string) as client:
+        db = client[DATABASE]
+        collection = db[collection_name]
+        max_sequence_number_doc = collection.find_one(
+            {}, sort=[("sequence_number", -1)]
+        )
+
+        if max_sequence_number_doc:
+            max_sequence_number = max_sequence_number_doc["sequence_number"]
+        else:
+            max_sequence_number = 0
+
+        return max_sequence_number
+
+
+def print_sync_status():
+    for collection_name in SYNC_COLLECTIONS:
+        max_sequence_number = get_max_sequence_number(collection_name)
+        print(f"Max sequence number for {collection_name}: {max_sequence_number}")
+
 
 def persist(collection: str, _id: str, data: dict):
     """
     Persist data to the database.
     """
-    # Using the 'with' statement for MongoDB client to ensure proper resource management
-    with MongoClient("mongodb://localhost:27017/") as client:
-        # Access the 'floriday' database
-        db = client["floriday"]
-
-        # Access the specified collection
+    print(f"Persisting data to {collection} on {mongodb_connection_string}...")
+    with MongoClient(mongodb_connection_string) as client:
+        db = client[DATABASE]
         coll = db[collection]
-
-        # Insert or update the document in the collection
         coll.update_one({"_id": _id}, {"$set": data}, upsert=True)

tests/test_upload_files_to_vine.py

@@ -39,7 +39,7 @@ def teardown_integration():
 
 def test_can_run_script():
     source_dir = "tests/data/"
-    cp = subprocess.run(["floridayvine", "upload", source_dir])
+    cp = subprocess.run(["floridayvine", "upload", source_dir, bucket])
     assert cp.returncode == 0
 
     file_count = count_files(source_dir)

work/backlog.md

@@ -10,13 +10,6 @@ Containers are pushed to <ghcr.io/serraict/vine-floriday-adapter>.
 ## Next
 
 * Goal: have a decent developer experience
-  * use the api to implement a generic sync implementation
-    * sync organizations from seq nr = 0
-    * sync from a certain number N
-      * sync from 0 up until N
-      * sync from N
-    * Data can be stored locally
-    * Mongo DB as part of this compose file makes more sense. Use serra-vine network.
   * learn about typer wrt exceptions and exit codes
   * document versioning strategy: what will we do when the api updates
   * organize typer commands into submodules