add workaround to ledger

Author: ryanraaschCDC

cfa/dataops/catalog.py

@@ -14,6 +14,7 @@
 import pandas as pd
 import polars as pl
 import tomli
+from azure.core.exceptions import ResourceExistsError
 from azure.identity import ManagedIdentityCredential
 from cfa.cloudops.blob_helpers import (
     read_blob_stream,
@@ -544,14 +545,31 @@ def ledger_entry(self, action: str) -> None:
             "action": action,
         }
         log_data = (json.dumps(log_entry) + "\n").encode("utf-8")
-        write_blob_stream(  # TODO: make this a streaming write to a single file (one per day parsed from get_timestamp())
-            data=log_data,
-            blob_url=f"{self.ledger_location['prefix']}/{get_date()}.jsonl",
-            account_name=self.ledger_location["account"],
-            container_name=self.ledger_location["container"],
-            append_blob=True,
-            overwrite=False,
-        )
+        ledger_path = f"{self.ledger_location['prefix']}/{get_date()}.jsonl"
+        try:
+            write_blob_stream(
+                data=log_data,
+                blob_url=ledger_path,
+                account_name=self.ledger_location["account"],
+                container_name=self.ledger_location["container"],
+                append_blob=True,
+                overwrite=False,
+            )
+        except ResourceExistsError as e:
+            # Azure Append Blobs have a 50,000 block limit; rotate to a new
+            # timestamped file when the daily file is full.
+            if "BlockCountExceedsLimit" in str(e):
+                rotated_path = f"{self.ledger_location['prefix']}/{get_date()}_{get_timestamp()}.jsonl"
+                write_blob_stream(
+                    data=log_data,
+                    blob_url=rotated_path,
+                    account_name=self.ledger_location["account"],
+                    container_name=self.ledger_location["container"],
+                    append_blob=True,
+                    overwrite=False,
+                )
+            else:
+                raise
 
     def save_dataframe(
         self,