Merge pull request #60 from CDCgov/rr-45-mocktest-data

Rr 45 mocktest data

Author: ryanraaschCDC

cfa/dataops/catalog.py

@@ -702,3 +702,80 @@ def dict_to_sn(d: Any, defaults: dict = None, ns: str = "") -> SimpleNamespace:
 datacat.__setattr__("__namespace_list__", dataset_namespaces)
 reportcat = SimpleNamespace(**combined_reports_dict)
 reportcat.__setattr__("__namespace_list__", report_namespaces)
+
+
+def _attach_schema_mock_functions(
+    datacat: SimpleNamespace, catalogs: list
+) -> None:
+    """Recursively walk the datacat namespace and attach mock_data functions to
+    the extract and load BlobEndpoints of each DatasetEndpoint, sourced from
+    a schema module co-located with the dataset.
+
+    The schema module is expected to live at:
+        {catalog_namespace}.{catalog_name}.datasets.{team_path}.schemas.{dataset_name}
+
+    where ``team_path`` is the namespace path segment(s) between the catalog
+    name and the dataset name (for example, ``stf`` in
+    ``public.stf.nhsn_hrd_prelim``), and ``dataset_name`` is the dataset's
+    namespace name (for example, ``nhsn_hrd_prelim``).
+
+    The schema module should define one or both of:
+        - extract_mock_data() -> pd.DataFrame
+        - load_mock_data() -> pd.DataFrame
+
+    These are then accessible as:
+        datacat.<catalog>.<team_path_segments>.<dataset>.extract.mock_data()
+        datacat.<catalog>.<team_path_segments>.<dataset>.load.mock_data()
+
+    Args:
+        datacat (SimpleNamespace): the top-level datacat namespace
+        catalogs (list): list of (catalog_namespace, catalog_name, catalog_path)
+            tuples from get_all_catalogs()
+    """
+
+    def _walk(ns: SimpleNamespace) -> None:
+        for val in vars(ns).values():
+            if isinstance(val, DatasetEndpoint):
+                # __ns_str__ is e.g. "public.stf.nhsn_hrd_prelim";
+                # the last segment is used as the schema module name
+                dataset_name = val.__ns_str__.split(".")[-1]
+                for cns, cat_name, _ in catalogs:
+                    # __ns_str__ is e.g. "public.stf.nhsn_hrd_prelim"
+                    # strip cat_name prefix -> "stf.nhsn_hrd_prelim"
+                    # then split into team ("stf") and dataset ("nhsn_hrd_prelim")
+                    # so the schema lives at: datasets.stf.schemas.nhsn_hrd_prelim
+                    ns_within_datasets = val.__ns_str__.removeprefix(
+                        f"{cat_name}."
+                    )
+                    ns_parts = ns_within_datasets.rsplit(".", 1)
+                    team_path = ns_parts[0] if len(ns_parts) > 1 else ""
+                    schema_mod_path = (
+                        f"{cns}.{cat_name}.datasets"
+                        f".{team_path}.schemas.{dataset_name}"
+                        if team_path
+                        else f"{cns}.{cat_name}.datasets.schemas.{dataset_name}"
+                    )
+                    try:
+                        mod = import_module(schema_mod_path)
+                    except ModuleNotFoundError:
+                        # No schema module for this dataset — skip silently
+                        continue
+                    # Attach to the already-existing BlobEndpoint on .extract / .load
+                    for stage, func_name in [
+                        ("extract", "extract_mock_data"),
+                        ("load", "load_mock_data"),
+                    ]:
+                        if hasattr(val, stage):
+                            if hasattr(mod, func_name):
+                                setattr(
+                                    getattr(val, stage),
+                                    "mock_data",
+                                    getattr(mod, func_name),
+                                )
+            elif isinstance(val, SimpleNamespace):
+                _walk(val)
+
+    _walk(datacat)
+
+
+_attach_schema_mock_functions(datacat, all_catalogs)

changelog.md

@@ -7,11 +7,17 @@ and this project adheres to [Calendar Versioning](https://calver.org/).
 The versioning pattern is `YYYY.MM.DD.micro(a/b/{none if release})
 
 ---
+## [2026.03.02.0]
+
+### Added
+
+- Added the ability to pull in mock data defined in a catalog.
+
 ## [2026.02.04.0]
 
 ### Added
 
-- Added capability to get_dataframe for reference datasets
+- Added capability to get_dataframe for reference datasets.
 
 
 ## [2026.01.05.0a]

docs/data_developer_guide.md

@@ -168,21 +168,24 @@ load_schema = pa.DataFrameSchema({
     "category": pa.Column(str)
 })
 
-# Add synthetic data generation for testing
-def raw_synthetic_data(n_rows: int = 100) -> pd.DataFrame:
-    return pd.DataFrame({
+# Add mock data generation for testing
+# prefix with 'extract' or 'load'
+def extract_mock_data(output="pandas", size=10) -> pd.DataFrame|pl.DataFrame:
+    data = {
             "date": pd.date_range(
                 start="2023-01-01",
-                periods=n_rows,
+                periods=size,
                 tz='UTC'
             ),
-            "value": np.random.uniform(1, 100, n_rows),
-            "category": np.random.choice(['A', 'B', 'C'], n_rows)
-    })
+            "value": np.random.uniform(1, 100, size),
+            "category": np.random.choice(['A', 'B', 'C'], size)
+    }
+    df = pd.DataFrame(data)
+    return df if output == "pandas" or output == "pd" else pl.from_pandas(df)
 
 # Validate synthetic data matches schema
 if __name__ == "__main__":
-    test_df = generate_synthetic_data()
+    test_df = extract_mock_data()
     extract_schema.validate(test_df)
 ```
 

pyproject.toml

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