Cdc-ap66 adding save methods and ledger updates (#32)

* new save methods and test

* version bump and making ledger an append type

* test update

* updating how save_dir_to_blob works

Author: cdc-ap66

cfa/dataops/catalog.py

@@ -8,7 +8,6 @@
 from io import BytesIO
 from types import SimpleNamespace
 from typing import Any, List, Sequence
-from uuid import uuid4
 
 import pandas as pd
 import polars as pl
@@ -28,7 +27,7 @@
     ValidationError,
 )
 from .reporting.catalog import report_dict_to_sn
-from .utils import get_dataset_dot_path, get_timestamp, get_user
+from .utils import get_dataset_dot_path, get_date, get_timestamp, get_user
 
 _here = os.path.abspath(os.path.dirname(__file__))
 _config = ConfigParser()
@@ -184,6 +183,7 @@ def write_blob(
         file_buffer: bytes | Sequence[bytes],
         path_after_prefix: str,
         auto_version: bool = False,
+        append: bool = False,
     ) -> None:
         """For writing file buffers to blob storage. Remember to include
         the a version to the path (i.e., {version}/{file}) or use
@@ -195,8 +195,9 @@ def write_blob(
             file_buffer (bytes or List[bytes]): the file buffer or list of buffers
             path_under_prefix (str): everything beyond the prefix
             auto_version (bool, optional): whether to automatically version
+            append (bool, optional): whether to append to existing file (only for single file writes).
         """
-        if auto_version:
+        if auto_version and not append:
             path_after_prefix = (
                 f"{get_timestamp()}/{path_after_prefix.lstrip('/')}"
             )
@@ -206,7 +207,7 @@ def write_blob(
             file_buffer = [file_buffer]
         total_partitions = len(file_buffer)
         for idx, fb_i in enumerate(file_buffer):
-            if total_partitions > 1:
+            if total_partitions > 1 and not append:
                 url_parts = os.path.splitext(full_path)
                 auto_full_path = f"{url_parts[0]}_{str(idx).zfill(len(str(total_partitions)))}{url_parts[1]}"
             else:
@@ -216,6 +217,7 @@ def write_blob(
                 blob_url=auto_full_path,
                 account_name=self.account,
                 container_name=self.container,
+                append_blob=append,
             )
         self.ledger_entry(action="write")
         # print(f"file written to: {full_path}")
@@ -383,11 +385,154 @@ def ledger_entry(self, action: str) -> None:
         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_timestamp()}_{uuid4().hex}.jsonl",
+            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,
         )
 
+    def save_dataframe(
+        self,
+        df: pd.DataFrame | pl.DataFrame,
+        path_after_prefix: str,
+        file_format: str = "parquet",
+        auto_version: bool = False,
+    ) -> None:
+        """Save a dataframe to the blob endpoint
+
+        Args:
+            df (pd.DataFrame | pl.DataFrame): the dataframe to save
+            path_after_prefix (str): the path after the prefix to save to
+            file_format (str, optional): the file format to save as.
+            Defaults to "parquet".
+            auto_version (bool, optional): whether to automatically version
+            the data. Defaults to True.
+            partition_cols (List[str], optional): columns to partition by
+            when saving. Defaults to None.
+        """
+        if file_format not in ["parquet", "csv", "json", "jsonl"]:
+            raise ValueError(
+                f"File format {file_format} not supported. Use 'parquet', 'csv', 'json', or 'jsonl'."
+            )
+        if file_format in ["json", "jsonl"] and path_after_prefix.endswith(
+            ".json"
+        ):
+            path_after_prefix = path_after_prefix[:-5] + ".jsonl"
+            print("Changing file extension to .jsonl for line-delimited JSON.")
+        if isinstance(df, pd.DataFrame):
+            if file_format == "parquet":
+                pq_bytes = df.to_parquet(index=False, compression="snappy")
+                self.write_blob(
+                    file_buffer=pq_bytes,
+                    path_after_prefix=path_after_prefix
+                    if path_after_prefix.endswith(".parquet")
+                    else path_after_prefix + ".parquet",
+                    auto_version=auto_version,
+                )
+            elif file_format == "csv":
+                csv_bytes = df.to_csv(index=False).encode("utf-8")
+                self.write_blob(
+                    file_buffer=csv_bytes,
+                    path_after_prefix=path_after_prefix
+                    if path_after_prefix.endswith(".csv")
+                    else path_after_prefix + ".csv",
+                    auto_version=auto_version,
+                )
+            elif file_format in ["json", "jsonl"]:
+                json_bytes = df.to_json(orient="records", lines=True).encode(
+                    "utf-8"
+                )
+                self.write_blob(
+                    file_buffer=json_bytes,
+                    path_after_prefix=path_after_prefix
+                    if path_after_prefix.endswith(".jsonl")
+                    else path_after_prefix + ".jsonl",
+                    auto_version=auto_version,
+                )
+        elif isinstance(df, pl.DataFrame):
+            if file_format == "parquet":
+                buffer = BytesIO()
+                df.write_parquet(buffer, compression="snappy")
+                pq_bytes = buffer.getvalue()
+                self.write_blob(
+                    file_buffer=pq_bytes,
+                    path_after_prefix=path_after_prefix
+                    if path_after_prefix.endswith(".parquet")
+                    else path_after_prefix + ".parquet",
+                    auto_version=auto_version,
+                )
+            elif file_format == "csv":
+                csv_bytes = df.write_csv().encode("utf-8")
+                self.write_blob(
+                    file_buffer=csv_bytes,
+                    path_after_prefix=path_after_prefix
+                    if path_after_prefix.endswith(".csv")
+                    else path_after_prefix + ".csv",
+                    auto_version=auto_version,
+                )
+            elif file_format in ["json", "jsonl"]:
+                json_bytes = df.write_ndjson().encode("utf-8")
+                self.write_blob(
+                    file_buffer=json_bytes,
+                    path_after_prefix=path_after_prefix
+                    if path_after_prefix.endswith(".jsonl")
+                    else path_after_prefix + ".jsonl",
+                    auto_version=auto_version,
+                )
+
+    def save_file_to_blob(
+        self,
+        file_path: str,
+        path_after_prefix: str,
+        auto_version: bool = False,
+    ) -> None:
+        """Save a local file to the blob endpoint
+
+        Args:
+            file_path (str): the local file path to save
+            path_after_prefix (str): the path after the prefix to save to
+            auto_version (bool, optional): whether to automatically version
+            the data. Defaults to False.
+        """
+        if not os.path.isfile(file_path):
+            raise ValueError(f"File {file_path} does not exist.")
+        with open(file_path, "rb") as f:
+            file_bytes = f.read()
+        self.write_blob(
+            file_buffer=file_bytes,
+            path_after_prefix=path_after_prefix,
+            auto_version=auto_version,
+        )
+
+    def save_dir_to_blob(
+        self,
+        dir_path: str,
+        path_after_prefix: str,
+        auto_version: bool = False,
+    ) -> None:
+        """Save a local directory to the blob endpoint
+
+        Args:
+            dir_path (str): the local directory path to save
+            path_after_prefix (str): the path after the prefix to save to
+            auto_version (bool, optional): whether to automatically version
+            the data. Defaults to False.
+        """
+        if not os.path.isdir(dir_path):
+            raise ValueError(f"Directory {dir_path} does not exist.")
+        for root, _, files in os.walk(dir_path):
+            for file in files:
+                file_path = os.path.join(root, file)
+                with open(file_path, "rb") as f:
+                    file_buffer = f.read()
+                rel_path = f"{'/'.join([i for i in os.path.split(root) if i])}/{'/'.join([i for i in os.path.split(file) if i])}"
+                self.write_blob(
+                    file_buffer=file_buffer,
+                    path_after_prefix=f"{path_after_prefix}/{rel_path}",
+                    auto_version=auto_version,
+                )
+
 
 def dict_to_sn(d: Any, defaults: dict = None, ns: str = "") -> SimpleNamespace:
     """Simple recursive namespace construction

cfa/dataops/utils.py

@@ -105,6 +105,15 @@ def get_timestamp(make_standard: bool = False) -> str:
     return datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
 
 
+def get_date() -> str:
+    """Get the current date in YYYY-MM-DD format.
+
+    Returns:
+        str: the current date as a string in YYYY-MM-DD format.
+    """
+    return datetime.now().strftime("%Y-%m-%d")
+
+
 def get_user() -> str:
     """Get the current system user
 

changelog.md

@@ -8,6 +8,16 @@ The versioning pattern is `YYYY.MM.DD.micro(a/b/{none if release})
 
 ---
 
+## [2025.10.03.0a]
+
+### Added
+
+- Save helper methods to BlobEndpoint
+
+### Updated
+
+- Made ledger append to date files
+
 ## [2025.09.25.0a]
 
 ### Updated

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "cfa.dataops"
-version = "2025.09.25.0a"
+version = "2025.10.03.0a"
 description = "Data cataloging, ETL, modeling, verification, and validation for CFA"
 authors = [
     { name = "Phil Rogers", email = "ap66@cdc.gov" },

tests/conftest.py

@@ -19,6 +19,8 @@ def mock_write_blob_stream(
         blob_url: str,
         account_name: str,
         container_name: str,
+        append_blob: bool = False,
+        overwrite: bool = True,
     ) -> None:
         return None