Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions cfa/dataops/catalog.py
Comment thread
ryanraaschCDC marked this conversation as resolved.
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""building a validated datasource namespace"""

import json
import logging
import os
import pkgutil
from collections.abc import Sequence
Expand Down Expand Up @@ -42,6 +43,10 @@
_config = ConfigParser()
_config.read(os.path.join(_here, "config.ini"))

logger = logging.getLogger(__name__)
if not logger.handlers:
logger.addHandler(logging.NullHandler())


def get_all_catalogs() -> list:
"""Get a list of all available dataops catalogs.
Expand All @@ -57,8 +62,10 @@ def get_all_catalogs() -> list:
for module_finder, modname, ispkg in pkgutil.iter_modules(catalog_pkg.__path__):
if ispkg:
catalogs.append((catalog_nspace, modname, module_finder.path))
except ModuleNotFoundError:
print(f"No catalogs exist in namespace {catalog_nspace}")
except ModuleNotFoundError as e:
if e.name != catalog_nspace:
raise
logger.warning("No catalogs exist in namespace %s", catalog_nspace)

return catalogs

Expand Down Expand Up @@ -340,6 +347,7 @@ def _get_version_blobs(
raise ValueError(
f"Version {version} not found in available versions: {available_versions}"
)
logger.info(f"Using version: {version}")
if print_version:
print(f"Using version: {version}")
if isinstance(version, list):
Expand Down Expand Up @@ -424,6 +432,7 @@ def get_dataframe(
output: Literal["pandas", "pd"] = "pandas",
version_spec: str | None = None,
selection: Literal["newest", "oldest"] = "newest",
print_version: bool = False,
) -> pd.DataFrame: ...
Comment thread
ryanraaschCDC marked this conversation as resolved.

@overload
Expand All @@ -432,6 +441,7 @@ def get_dataframe(
output: Literal["polars", "pl"],
version_spec: str | None = None,
selection: Literal["newest", "oldest"] = "newest",
print_version: bool = False,
) -> pl.DataFrame: ...
Comment thread
ryanraaschCDC marked this conversation as resolved.

@overload
Expand All @@ -440,13 +450,15 @@ def get_dataframe(
output: Literal["pl_lazy", "lazy"],
version_spec: str | None = None,
selection: Literal["newest", "oldest"] = "newest",
print_version: bool = False,
) -> pl.LazyFrame: ...
Comment thread
ryanraaschCDC marked this conversation as resolved.

def get_dataframe(
self,
output: Literal["pandas", "pd", "polars", "pl", "pl_lazy", "lazy"] = "pandas",
version_spec: str | None = None,
selection: Literal["newest", "oldest"] = "newest",
print_version: bool = False,
) -> pd.DataFrame | pl.DataFrame | pl.LazyFrame:
Comment thread
ryanraaschCDC marked this conversation as resolved.
"""Get the data as a pandas or polars dataframe

Expand All @@ -456,6 +468,7 @@ def get_dataframe(
version_spec (str, optional): the version of the data to get.
Defaults to "latest".
selection (Literal["newest", "oldest", "all"], optional): whether to get the newest, oldest, or all matching versions. Defaults to "newest".
print_version (bool, optional): whether to log the resolved version information. Defaults to False.
Comment thread
ryanraaschCDC marked this conversation as resolved.

Raises:
ValueError: if output is not one of
Expand All @@ -472,7 +485,7 @@ def get_dataframe(
)
# Fetch version blobs once and validate before deriving file extension.
version_blobs = self._get_version_blobs(
version_spec=version_spec, selection=selection
version_spec=version_spec, selection=selection, print_version=print_version
)
if not version_blobs:
raise ValueError(
Expand Down Expand Up @@ -627,7 +640,7 @@ def save_dataframe(
)
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.")
logger.info("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")
Expand Down
4 changes: 4 additions & 0 deletions changelog.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,10 @@ and this project adheres to [Calendar Versioning](https://calver.org/).
The versioning pattern is `YYYY.MM.DD.micro(a/b/{none if release})

---
## [2026.07.20.0]

- change most print statements to logging

## [2026.06.30.0]

- changed SimpleNamespace to CatalogNamespace for more helpful type hints
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "cfa.dataops"
version = "2026.06.30.0"
version = "2026.07.20.0"
description = "Data cataloging, ETL, modeling, verification, and validation for CFA"
authors = [
{ name = "Phil Rogers", email = "ap66@cdc.gov" },
Expand Down
20 changes: 9 additions & 11 deletions tests/test_blob_endpoint_save_methods.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,25 +128,23 @@ def test_save_pandas_dataframe_json(self, mocker, blob_endpoint, sample_pandas_d
assert isinstance(call_args[1]["file_buffer"], bytes)

def test_save_pandas_dataframe_json_auto_extension_fix(
self, mocker, blob_endpoint, sample_pandas_df, capsys
self, mocker, blob_endpoint, sample_pandas_df, caplog
):
"""Test that .json extension is automatically changed to .jsonl"""
mock_write = mocker.patch.object(blob_endpoint, "write_blob")

blob_endpoint.save_dataframe(
df=sample_pandas_df,
path_after_prefix="data/output.json",
file_format="json",
auto_version=False,
)
with caplog.at_level("INFO"):
blob_endpoint.save_dataframe(
df=sample_pandas_df,
path_after_prefix="data/output.json",
file_format="json",
auto_version=False,
)

mock_write.assert_called_once()
call_args = mock_write.call_args
assert call_args[1]["path_after_prefix"] == "data/output.jsonl"

# Check that warning message was printed
captured = capsys.readouterr()
assert "Changing file extension to .jsonl" in captured.out
assert "Changing file extension to .jsonl" in caplog.text

def test_save_polars_dataframe_parquet(
self, mocker, blob_endpoint, sample_polars_df
Expand Down