From 590f455acfb6b872af3875458ee607a4b9adb084 Mon Sep 17 00:00:00 2001 From: Ryan Raasch Date: Tue, 12 May 2026 17:23:31 +0000 Subject: [PATCH 1/4] disable ledger --- cfa/dataops/catalog.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/cfa/dataops/catalog.py b/cfa/dataops/catalog.py index 279752a..1ca3479 100644 --- a/cfa/dataops/catalog.py +++ b/cfa/dataops/catalog.py @@ -217,7 +217,7 @@ def write_blob( container_name=self.container, append_blob=append, ) - self.ledger_entry(action="write") + # self.ledger_entry(action="write") # print(f"file written to: {full_path}") def read_blobs( @@ -242,7 +242,7 @@ def read_blobs( ) for i in blobs ] - self.ledger_entry(action="read") + # self.ledger_entry(action="read") return blob_bytes def read_csv(self, suffix: str) -> pd.DataFrame: @@ -252,7 +252,7 @@ def read_csv(self, suffix: str) -> pd.DataFrame: container_name=self.container, ) df = pd.read_csv(blob) - self.ledger_entry(action="read") + # self.ledger_entry(action="read") return df def get_versions(self) -> list: @@ -372,8 +372,8 @@ def download_version_to_local( with open(local_file_path, "wb") as f: f.write(file_bytes) written = True - if written: - self.ledger_entry(action="read") + # if written: + # self.ledger_entry(action="read") return written @overload @@ -446,7 +446,7 @@ def get_dataframe( credential=ManagedIdentityCredential() ), ) - self.ledger_entry(action="read") + # self.ledger_entry(action="read") return df elif file_ext == "csv": path = str(PurePosixPath(name).parent / f"*.{file_ext}") @@ -459,7 +459,7 @@ def get_dataframe( credential=ManagedIdentityCredential() ), ) - self.ledger_entry(action="read") + ##self.ledger_entry(action="read") return df elif file_ext == "ndjson" or file_ext == "jsonl": path = str(PurePosixPath(name).parent / f"*.{file_ext}") @@ -472,7 +472,7 @@ def get_dataframe( credential=ManagedIdentityCredential() ), ) - self.ledger_entry(action="read") + ##self.ledger_entry(action="read") return df else: raise ValueError(f"Lazy loading not supported for {file_ext} files.") From 12f5ad385c04d30751e4ee59afdc59a81f0be72a Mon Sep 17 00:00:00 2001 From: Ryan Raasch Date: Tue, 12 May 2026 17:25:53 +0000 Subject: [PATCH 2/4] update changelog, pyproject --- changelog.md | 6 ++++++ pyproject.toml | 2 +- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/changelog.md b/changelog.md index 2445091..71e6999 100644 --- a/changelog.md +++ b/changelog.md @@ -7,6 +7,12 @@ and this project adheres to [Calendar Versioning](https://calver.org/). The versioning pattern is `YYYY.MM.DD.micro(a/b/{none if release}) --- +## [2026.05.12.0] + +### Fixed + +- removed the ledger reads and write temporarily + ## [2026.04.27.0] ### Fixed diff --git a/pyproject.toml b/pyproject.toml index afb80c2..5d5705f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "cfa.dataops" -version = "2026.04.27.0" +version = "2026.05.12.0" description = "Data cataloging, ETL, modeling, verification, and validation for CFA" authors = [ { name = "Phil Rogers", email = "ap66@cdc.gov" }, From 65cd9d6ef54ff10cb14088bca429c98aae89f1e1 Mon Sep 17 00:00:00 2001 From: Ryan Raasch Date: Tue, 12 May 2026 17:48:34 +0000 Subject: [PATCH 3/4] add workaround to ledger --- cfa/dataops/catalog.py | 34 ++++++++++++++++++++++++++-------- 1 file changed, 26 insertions(+), 8 deletions(-) diff --git a/cfa/dataops/catalog.py b/cfa/dataops/catalog.py index 1ca3479..e263b41 100644 --- a/cfa/dataops/catalog.py +++ b/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, From 68dfa9ee0c8ac4b8e0f4f5b06e7a3803c6ab4f81 Mon Sep 17 00:00:00 2001 From: Ryan Raasch Date: Tue, 12 May 2026 18:21:40 +0000 Subject: [PATCH 4/4] remove copilot ledger code --- cfa/dataops/catalog.py | 34 +++++++++------------------------- 1 file changed, 9 insertions(+), 25 deletions(-) diff --git a/cfa/dataops/catalog.py b/cfa/dataops/catalog.py index e263b41..04b992c 100644 --- a/cfa/dataops/catalog.py +++ b/cfa/dataops/catalog.py @@ -14,7 +14,6 @@ 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, @@ -546,30 +545,15 @@ def ledger_entry(self, action: str) -> None: } log_data = (json.dumps(log_entry) + "\n").encode("utf-8") 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 + + 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, + ) def save_dataframe( self,